diff --git a/AGENTIC-AUDIT-REPORT.md b/AGENTIC-AUDIT-REPORT.md new file mode 100644 index 000000000..7cae5b579 --- /dev/null +++ b/AGENTIC-AUDIT-REPORT.md @@ -0,0 +1,125 @@ +# AI 助手（Agentic）体系体检报告 + +> 审计日期：2026-10-20 · 方法：4 阶段多智能体审计（8 读图 → 7 维度深审 → 每发现 3 镜头对抗验证 → 完备性批评） +> 规模：220 个 agent · 4,538 次工具调用 · 约 1,100 万 tokens · 81 分钟 +> 结论口径：**62 条确认**（15 high / 33 medium / 14 low），6 条被对抗验证否决，13 条因 API 限流票数不全（均已获 ≥2 票通过，结论可信但置信略降） + +--- + +## 一、总评（直接回答三个问题） + +### 1. 健不健全？——**骨架健全，多处未达生产水位** + +分层设计、依赖方向、端口-适配器分工全部符合仓库既有模式且干净。但有 15 个 high 级缺陷集中在**韧性**（幻觉工具名/慢消费者杀死整轮会话）、**安全**（SSRF 探测面、API Key 明文落库、提示注入零防御）、**契约**（OpenAI 兼容承诺与实现背离）三块——这些在单用户开发场景无感，多租户生产场景会集中引爆。 + +### 2. 全不全面？——**主体全面，但有五块"版图外"区域零审计** + +12 工具 + 双运行时 + 会话/附件/动作管理 + 前端全套 + 41 测试的主体覆盖完整。但批评员发现五个**边界组成部分**从未被任何维度触及（见第六节），最典型的是 **MCP 外供通道**（平台把 API 以 MCP 工具暴露给外部 AI 客户端）——这是 AI 助手的"第三条暴露面"，完全零审计。 + +### 3. 是最佳实践吗？——**局部达标，体系不是** + +**已达最佳实践水位的**：租户上下文强制注入（不信任模型参数）、AgenticToolResult 统一信封、写操作 10 分钟人审门、前端 SWR 不闪/XSS 消毒/竞态双令牌/千例对话语料、契约测试纪律。 + +**低于最佳实践的五个体系级差距**（均有 high 级证据）： +1. 供应商抽象不完整——换 Anthropic 等非 OpenAI 兼容供应商 = 静默砍掉 95% 工具面，且系统提示词仍向模型许诺全部能力 +2. 上下文管理无 token 预算——按条数截窗、字符/4 估算用量、analytics 结果 5000 点无截断直接注入 +3. Agent 循环终止 fail-hard——超 8 轮直接报错杀整轮，无重复调用检测、无末轮去工具强制收尾 +4. 零指标埋点——LLM 延迟/工具成功率/token/队列深度全部不可度量，prometheus 端点空挂 +5. 多副本语义未设计——进程内缓存无 TTL、失效仅本节点、限流按节点而非全局 + +--- + +## 二、High 级缺陷（15 条，对抗验证通过） + +### 韧性组（agent 循环） +| # | 缺陷 | 位置 | +|---|---|---| +| H1 | **幻觉工具名 `Mono.error` 杀死整轮会话**——未知工具应返回 tool-error 让模型自愈（轮数上限已有兜底） | `OpenAiCompatibleAgenticRuntime.java:456` | +| H2 | **慢 SSE 消费者被硬性判死**——`requestedFromDownstream()==0` 即终止生成并落库 FAILED，零缓冲容忍 | `OpenAiCompatibleAgenticRuntime.java:271` | +| H3 | **双通道工具注册造成 Provider 级能力断层**——Spring AI 路径仅 3 类 @Tool（约 10 方法），60+ reactive 工具整体缺席；桥接代码在 L463-466 已有反向先例可对称复用 | `ChatClientConfig.java:106` | + +### 安全组 +| # | 缺陷 | 位置 | +|---|---|---| +| H4 | **连通性探测成 SSRF 通道**——草稿探测接受任意 baseUrl；providerId+自定义 baseUrl 组合会把存量 API Key 发往攻击者主机。需 https-only + 私网/环回段拒绝（解析后校验防 DNS rebinding）+ 改 baseUrl 必须同时换 key | `AgenticConnectivityCheckServiceImpl.java:160` | +| H5 | **LLM API Key 明文落库、无 TTL 常驻堆缓存**——静态加密完全缺失；建议信封加密（AES-GCM+主密钥）、缓存存脱敏副本 | `R2dbcModelProviderStore.java:46` | +| H6 | **提示注入零防御**——附件元数据无定界符直接拼 system prompt，工具结果回流无消毒。需注入防御段 + 随机化定界符 + 已知注入模式过滤 | `ChatClientConfig.java:53` | + +### 契约组 +| # | 缺陷 | 位置 | +|---|---|---| +| H7 | **非流式响应输出 camelCase 字段名**（finishReason/promptTokens）——背离自称的 OpenAI 兼容契约；请求的 max_tokens 被静默忽略 | `ChatCompletionResponseVO.java:93` | +| H8 | **conversationId 运行时必填但 OpenAPI 声明可选**——标准 OpenAI 客户端每次请求必失败；文档与实现互相矛盾 | `AgenticChatRequestPreparer.java:178` | + +### 可观测组 +| # | 缺陷 | 位置 | +|---|---|---| +| H9 | **agentic 模块零指标**——五组必备指标（信号量水位/轮次 Timer/工具 Counter/token 用量/队列深度）全缺 | `AgenticChatServiceImpl.java:173` | +| H10 | **trace 贯穿断裂**——requestId 从未写入 MDC（日志关联槽恒空），insight RPC 不携带关联 ID，跨服务排障断链 | `RequestIdWebFilter.java:103` | + +### 前端组 +| # | 缺陷 | 位置 | +|---|---|---| +| H11 | **Composer 回车在 IME 输入中途提交**——中日韩输入法确认候选词的 Enter 被当发送，中文用户无法正常输入 | `AgenticComposer.vue:80` | + +### 最佳实践组 +| # | 缺陷 | 位置 | +|---|---|---| +| H12 | 同 H3（供应商维度视角） | — | +| H13 | **上下文无 token 预算**（同总评差距 2） | `AgenticProperties.java:64` | +| H14 | **工具 schema 手写内联 JSON 且 7 个 analytics 工具的 series/window 是裸 `{"type":"object"}`**——schema 字面量与代码校验常数平行维护必然漂移；QUERY_LATEST 已证明可填全 | `ReactiveAgenticToolRegistry.java:325` | +| H15 | 同 H1（最佳实践维度视角） | — | + +--- + +## 三、Medium 级主题（33 条，按主题归并） + +- **架构**：1180 行注册表 8 组复制粘贴工厂/接口/解析器（一个泛型工厂可删约 500 行）；telescoping 构造器静默丢工具；`tools()` 每轮全量重建且排序被 `Map.copyOf` 丢弃；db 适配器反向注入 service 层执行文件系统副作用（分层倒置）；MessageController 绕过 service 直连 repository 且无分页；ChatClientFactory 四张缓存无 TTL、fallback 客户端永久粘滞 +- **正确性**：软删会话墓碑永久毒化 conversationId（唯一约束冲突抛裸异常）；session_ext 并发读-合-写丢失更新（无锁无 CAS）；阻塞文件操作跑在 Netty 事件循环；TIMESTAMPTZ 绑定 UTC-naive LocalDateTime（时区漂移）；附件 TOCTOU 孤儿文件 +- **安全**：连通性缓存无淘汰且 key 由客户端材料构成（内存 DoS）；租户缺模型配置时静默回退平台全局 Key 且该路径不限流；权限动词错位（批准高危写操作仅需 `action:list`，上传/聊天复用 list/get 动词）；附件无 MIME 白名单无租户配额 +- **契约**：请求 messages 数组声明为完整上下文但只取最后一条 user；三种分页风格并存；AttachmentVO 泄露服务端存储路径；SessionVO 声明永远返回 null 的字段 +- **可观测**：provider 真实 usage 被 includeUsage 请求后丢弃（全靠估算且系统性偏低）；限流半成品（无超时无队列上限，未知名模型完全绕过）；SSE 无心跳无事件名、失败恒 HTTP 200 收尾 +- **前端**：流式热路径 O(n²)（每 delta 全量 markdown 重解析+DOMPurify）；mock 引擎整段输出不模拟逐 token 流；会话列表硬上限 50 无翻页（更早会话 UI 不可达）；停靠模式下拉与工作台栏数据口径分叉 +- **最佳实践**：数值参数 `10.0`/`1e3` 形态抛 NumberFormatException；工具注册表与 provider 格式定义每请求重建 + +## 四、Low 级（14 条） + +5 个注入未用的工具参数（重构残留）；OpenAI 客户端三处复制（运行时每请求新建无连接池复用）；工具异常 message 原样回灌模型（内部拓扑泄露）；provider 路由嵌套不一致；运维日志缺租户归属；fetch 通道 401 处理与 axios 拦截器漂移；失败回合无重试按钮；等——详见附录数据。 + +--- + +## 五、做得好的（必须说） + +1. **租户隔离是全审计最强项**：每个工具从 ToolContext 强制提取 tenantId（不信任模型参数）、InsightDispatcher 二次校验、DB 层 WHERE 全带 tenant_id、写点值 10 分钟人审门全程租户校验 +2. **统一信封设计**：AgenticToolResult{success,code,message,data,visualizations} 六态 code 枚举，已注册工具的错误被 tracing 层降级为 tool-result 不炸循环（讽刺的是未知工具名反而炸——H1 正是没对齐这个自己的好模式） +3. **前端工程质量高于社区平均**：SWR 缓存播种+尾部对齐合并（无重载闪烁）、幽灵占位、reduced-motion 双处生效、多宽度适配、DOMPurify 全量消毒、竞态双令牌、1000+ 用例对话语料 +4. **仓库纪律**：禁 @PathVariable 全模块合规（grep 验证）、RFC 9457 problem+json 统一错误、密钥 VO write-only + LogSanitizer 达标、消息索引 FOR UPDATE 正确串行化、action 审批 CAS 竞态安全 + +--- + +## 六、完备性缺口（批评员裁定，二轮审计候选） + +1. **MCP 外供通道**（`dc3-common-auth` 的 McpManagementController/ReactiveOAuthMcpRuntimeServiceImpl + `openapi-agentic.json` 冻结快照 + dc3-cli mcp.ts）——第三条暴露面，且快照与 7 个 controller 实际契约的漂移**无任何守护** +2. **ActiveAgenticProfileConfig** 经 spring.factories 以 HIGHEST_PRECEDENCE 无条件激活 agentic profile 到任何宿主（含 center-single）——未审计 +3. **native/AOT 形态**：手写 JSON schema + ObjectMapper + MethodToolCallbackProvider 反射在 native 管线（真实发布形态）下行为从未验证 +4. **权限种子闭环**：全部 initdb/migrations grep 不到任何 agentic 资源码——@perm.can 引用的资源从何而来、默认谁能批准写动作，未定性（verb 错位是笔误还是有意无法判断） +5. **数据保留与隐私**：dc3_message（全量对话+工具轨迹）与 dc3_attachment 无保留期策略、无租户清理任务、无合规视角 +6. **多副本部署**：SSE 网关粘性、缓存跨节点失效、信号量按节点——多处仅"注释自认退化" +7. **Insight TSDB 租户隔离未下钻**（ReactiveTsdbStore 是否强制 tenant 分量）——租户隔离是硬规则，此处必须二轮实证 +8. 文档面（iot-dc3-docs 仓库）、可达性/i18n、mock 契约守护、工具层测试盲区（CommandTool/EventTool 恰是最弱校验处且零专测） + +## 七、修复路线建议 + +- **P0（发布前必须）**：H4 SSRF、H5 密钥加密、H1+H2 循环韧性（都是小改动大收益）、H11 IME（一行守卫） +- **P1（生产水位）**：H3 双通道桥接、H13 token 预算、H7/H8 契约对齐、H9+H10 指标+trace、权限动词、附件白名单+配额、限流补全 +- **P2（工程债）**：注册表泛型化重构（-500 行）、分页统一、缓存 TTL、前端 O(n²) 批量合并、MessageController 分页 +- **二轮审计**：第六节 8 项（MCP 通道优先） + +--- + +## 附：方法论与可信度说明 + +- 7 维度审查员各自独立读码（非仅依赖读图摘要），每条 finding 带 file:line+代码证据+可落地建议 +- 每条发现经 3 镜头对抗验证（证据核对/影响评估/主动反驳），≥2/3 票保留；13 条因 API 限流仅 2 票（全部为 2 正票，无反对票） +- 6 条被否决项含一条高价值仲裁：审计员称"信号量许可取消时永久泄漏为最严重缺陷"，验证员否决，人工复核确认**否决正确**（`Flux.usingWhen` 三参重载在 cancel 路径同样执行 release；仅剩 acquire 阻塞中途被取消的极窄边缘） +- 完整结构化数据（62 条全文+8 张子系统地图+7 份维度总结+批评报告）：会话任务输出 `wegnwa0xu.output`，journal 见 `subagents/workflows/wf_3344a3b1-c53/journal.jsonl` diff --git a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveAttachmentStore.java b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveAttachmentStore.java index a33697c28..a739ea357 100644 --- a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveAttachmentStore.java +++ b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveAttachmentStore.java @@ -25,12 +25,25 @@ import reactor.core.publisher.Mono; /** Reactive persistence port for attachment metadata. */ public interface ReactiveAttachmentStore { - /** Save the attachment, inserting or updating as needed. */ + /** + * Save the attachment. The insert runs in one transaction with a live-session guard + * ({@code SELECT ... FOR UPDATE} on the owning session), so a save racing a concurrent + * session delete either lands before the delete or fails with a clear error instead of + * leaving a zombie row for a soft-deleted conversation. + */ Mono save(AttachmentBO attachment); /** Stream attachments matching the request. */ Flux list(String conversationId, RequestHeader.PrincipalHeader header); + /** + * Stream the stored file paths of the conversation's soft-deleted attachments. Used by the + * session delete orchestration to unlink files after the database delete commits; live + * attachments are excluded on purpose so a conversation revived after the delete keeps its + * newly uploaded files. + */ + Flux listDeletedFilePaths(String conversationId, RequestHeader.PrincipalHeader header); + /** Load the attachments for the given ids. */ Flux getByIds(Collection ids, RequestHeader.PrincipalHeader header); diff --git a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveSessionStore.java b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveSessionStore.java index 608852abb..06f7ced03 100644 --- a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveSessionStore.java +++ b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveSessionStore.java @@ -26,7 +26,15 @@ import reactor.core.publisher.Mono; /** Reactive persistence port for agentic conversation sessions. */ public interface ReactiveSessionStore { - /** Create or refresh the session and emit the stored state. */ + /** + * Create or refresh the session and emit the stored state. + * + *

Soft-deleted conversation ids are reusable: when the conversation's row is a soft-delete + * tombstone, {@code touch} revives that row as a brand-new session (title and extension reset, + * creation restarted) instead of failing on the natural-key uniqueness constraint, which must + * stay a full constraint because {@code dc3_message}, {@code dc3_attachment}, and + * {@code dc3_action} reference it with foreign keys. + */ Mono touch(String conversationId, SessionExt sessionExt, RequestHeader.PrincipalHeader header); /** Load the session for the request. */ @@ -40,14 +48,20 @@ public interface ReactiveSessionStore { java.util.List sort, RequestHeader.PrincipalHeader header); - /** Update one session and emit the updated row. */ + /** + * Update one session and emit the updated row. The read-merge-write sequence runs under a + * row lock inside one transaction, so concurrent updates merge instead of overwriting each + * other. Completes empty when the session does not exist. + */ Mono update( String conversationId, SessionExt sessionExt, String title, RequestHeader.PrincipalHeader header); /** - * Delete the session. Implementations also remove the session's stored attachment files - * best-effort after the database delete commits; file cleanup failures must not fail the - * delete. + * Soft-delete the session and its messages, attachments, and actions in one transaction, + * locking the session row first so concurrent attachment uploads serialize against the + * delete. Returns the number of deleted session rows (0 when already deleted); stored + * attachment file cleanup is the caller's responsibility, driven by + * {@link ReactiveAttachmentStore#listDeletedFilePaths}. */ Mono delete(String conversationId, RequestHeader.PrincipalHeader header); } diff --git a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleaner.java b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleaner.java index 7af0cddfd..dc76c3b12 100644 --- a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleaner.java +++ b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleaner.java @@ -23,33 +23,49 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.util.Collection; -import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; +import reactor.core.scheduler.Schedulers; /** * Best-effort filesystem cleanup for stored conversation attachments. * - *

Owns the attachment storage-root policy so upload path resolution and delete-time - * file unlinking share one root and one boundary guard and can never drift apart. All - * operations are warn-only: a missing, undeletable, or out-of-root file must never fail - * the database side of an attachment or session delete. + *

Owns the attachment storage-root policy so upload path resolution and delete-time file + * unlinking share one root and one boundary guard and can never drift apart. All operations are + * warn-only: a missing, undeletable, or out-of-root file must never fail the database side of an + * attachment or session delete. The blocking {@link Files} unlink always runs on the injected + * {@link Scheduler} (bounded elastic in production), so no caller can execute file deletion on a + * Reactor Netty event loop. * * @author pnoker * @since 2026.9.22 */ @Slf4j @Component -@RequiredArgsConstructor public class AttachmentFileCleaner { private final AgenticProperties properties; + private final Scheduler unlinkScheduler; private Path storageRoot; + /** Spring wiring: unlinks run on the shared bounded-elastic scheduler. */ + @Autowired + public AttachmentFileCleaner(AgenticProperties properties) { + this(properties, Schedulers.boundedElastic()); + } + + /** Test and alternate-wiring path: unlinks run on the given scheduler. */ + AttachmentFileCleaner(AgenticProperties properties, Scheduler unlinkScheduler) { + this.properties = properties; + this.unlinkScheduler = unlinkScheduler; + } + @PostConstruct void initializeStorageRoot() { storageRoot = storageRoot(); @@ -63,24 +79,26 @@ public class AttachmentFileCleaner { /** Unlink one stored attachment file, resolving it and enforcing the storage-root boundary. */ public Mono unlink(String storedFilePath) { if (StringUtils.isBlank(storedFilePath)) return Mono.empty(); - return Mono.fromRunnable(() -> { - try { - Files.deleteIfExists(resolveWithinRoot(storedFilePath)); - } catch (Exception exception) { - log.warn("Attachment file cleanup failed, file={}", storedFilePath, exception); - } - }); + return Mono.fromRunnable(() -> { + try { + Files.deleteIfExists(resolveWithinRoot(storedFilePath)); + } catch (Exception exception) { + log.warn("Attachment file cleanup failed, file={}", storedFilePath, exception); + } + }) + .subscribeOn(unlinkScheduler); } /** Unlink one already-resolved attachment file, used for upload rollback of just-created paths. */ Mono unlink(Path path) { - return Mono.fromRunnable(() -> { - try { - Files.deleteIfExists(path); - } catch (Exception exception) { - log.warn("Attachment file cleanup failed, file={}", path, exception); - } - }); + return Mono.fromRunnable(() -> { + try { + Files.deleteIfExists(path); + } catch (Exception exception) { + log.warn("Attachment file cleanup failed, file={}", path, exception); + } + }) + .subscribeOn(unlinkScheduler); } /** Unlink every stored attachment file in the collection, each best-effort. */ diff --git a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImpl.java b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImpl.java index 9e3735e91..7c58bd135 100644 --- a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImpl.java +++ b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImpl.java @@ -18,6 +18,7 @@ package io.github.pnoker.common.agentic.service.impl; import io.github.pnoker.common.agentic.entity.bo.SessionBO; import io.github.pnoker.common.agentic.entity.model.SessionExt; +import io.github.pnoker.common.agentic.repository.ReactiveAttachmentStore; import io.github.pnoker.common.agentic.repository.ReactiveSessionStore; import io.github.pnoker.common.agentic.service.SessionService; import io.github.pnoker.common.entity.common.RequestHeader; @@ -33,6 +34,8 @@ import reactor.core.publisher.Mono; public class SessionServiceImpl implements SessionService { private final ReactiveSessionStore sessionStore; + private final ReactiveAttachmentStore attachmentStore; + private final AttachmentFileCleaner attachmentFileCleaner; @Override public Mono touch(String conversationId, RequestHeader.PrincipalHeader header, SessionExt sessionExt) { @@ -62,6 +65,23 @@ public class SessionServiceImpl implements SessionService { @Override public Mono delete(String conversationId, RequestHeader.PrincipalHeader header) { - return sessionStore.delete(conversationId, header); + return sessionStore + .delete(conversationId, header) + .flatMap(deleted -> deleted > 0 + ? unlinkDeletedAttachmentFiles(conversationId, header).thenReturn(deleted) + : Mono.just(deleted)); + } + + /** + * Unlink the stored files of the conversation's soft-deleted attachments after the session + * delete commits. Runs strictly after the transaction so file cleanup can never roll the + * delete back, and stays best-effort: a missing or undeletable file never fails the committed + * delete. + */ + private Mono unlinkDeletedAttachmentFiles(String conversationId, RequestHeader.PrincipalHeader header) { + return attachmentStore + .listDeletedFilePaths(conversationId, header) + .collectList() + .flatMap(attachmentFileCleaner::unlinkAll); } } diff --git a/dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleanerTest.java b/dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleanerTest.java new file mode 100644 index 000000000..161b233e4 --- /dev/null +++ b/dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleanerTest.java @@ -0,0 +1,120 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package io.github.pnoker.common.agentic.service.impl; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.github.pnoker.common.agentic.config.AgenticProperties; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import reactor.core.scheduler.Scheduler; +import reactor.core.scheduler.Schedulers; + +/** + * Unit coverage of the attachment file cleaner's blocking-IO offload: the NIO unlink must run on + * the injected scheduler (bounded elastic in production) so no caller can execute file deletion + * on a Reactor Netty event loop, while the storage-root boundary guard and best-effort contract + * stay intact. + */ +class AttachmentFileCleanerTest { + + @TempDir + Path storageRoot; + + private final ConcurrentLinkedQueue unlinkThreads = new ConcurrentLinkedQueue<>(); + private ExecutorServiceHolder holder; + + private AttachmentFileCleaner cleaner; + + /** Keeps a reference to the probe executor so the test can shut it down. */ + private record ExecutorServiceHolder(java.util.concurrent.ExecutorService service) {} + + @BeforeEach + void setUp() { + AgenticProperties properties = new AgenticProperties(); + properties.setAttachmentStoragePath(storageRoot.toString()); + java.util.concurrent.ExecutorService probe = Executors.newSingleThreadExecutor(runnable -> { + Thread thread = new Thread(runnable, "probe-unlink"); + thread.setDaemon(true); + return thread; + }); + holder = new ExecutorServiceHolder(probe); + // The probe executor records the thread that actually performs each task, proving the + // cleaner offloads instead of running on the subscribing thread. + Executor instrumented = command -> probe.execute(() -> { + unlinkThreads.add(Thread.currentThread().getName()); + command.run(); + }); + Scheduler scheduler = Schedulers.fromExecutor(instrumented); + cleaner = new AttachmentFileCleaner(properties, scheduler); + cleaner.initializeStorageRoot(); + } + + @AfterEach + void tearDown() { + holder.service().shutdownNow(); + } + + @Test + void unlinkRunsTheBlockingDeletionOnTheSchedulerThread() throws Exception { + Path file = Files.writeString(storageRoot.resolve("note.txt"), "payload"); + String subscribingThread = Thread.currentThread().getName(); + + cleaner.unlink(file.toString()).block(); + + assertThat(Files.exists(file)).as("the file must be deleted").isFalse(); + assertThat(unlinkThreads) + .as("the unlink must run on the scheduler thread, not on the subscriber") + .singleElement() + .isNotEqualTo(subscribingThread) + .satisfies(thread -> assertThat(thread).startsWith("probe-unlink")); + } + + @Test + void unlinkAllRunsEveryDeletionOnTheSchedulerThread() throws Exception { + Path first = Files.writeString(storageRoot.resolve("a.txt"), "payload"); + Path second = Files.writeString(storageRoot.resolve("b.txt"), "payload"); + + cleaner.unlinkAll(List.of(first.toString(), second.toString(), " ")).block(); + + assertThat(Files.exists(first)).isFalse(); + assertThat(Files.exists(second)).isFalse(); + assertThat(unlinkThreads).as("each unlink must be offloaded").hasSize(2); + } + + @Test + void unlinkSkipsBlankPathsAndRefusesPathsOutsideTheStorageRoot() throws Exception { + Path outside = Files.createTempDirectory("dc3-cleaner-outside"); + Path escaped = Files.writeString(outside.resolve("escaped.txt"), "payload"); + + cleaner.unlink(" ").block(); + assertThat(unlinkThreads).as("blank paths must not schedule any work").isEmpty(); + + cleaner.unlink(escaped.toString()).block(); + assertThat(Files.exists(escaped)) + .as("the boundary guard must refuse out-of-root unlinks") + .isTrue(); + } +} diff --git a/dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImplTest.java b/dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImplTest.java index 47a35d176..a284ed58d 100644 --- a/dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImplTest.java +++ b/dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImplTest.java @@ -16,36 +16,61 @@ */ package io.github.pnoker.common.agentic.service.impl; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import io.github.pnoker.common.agentic.config.AgenticProperties; import io.github.pnoker.common.agentic.entity.bo.SessionBO; +import io.github.pnoker.common.agentic.repository.ReactiveAttachmentStore; import io.github.pnoker.common.agentic.repository.ReactiveSessionStore; import io.github.pnoker.common.entity.common.RequestHeader; import io.github.pnoker.db.core.page.OffsetPage; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.List; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; +/** + * Unit coverage of the session service delete orchestration: the service owns the post-commit + * attachment file cleanup (the store stays pure SQL), cleanup failures never fail the committed + * delete, and an already-deleted session skips cleanup entirely. + */ @ExtendWith(MockitoExtension.class) class SessionServiceImplTest { @Mock private ReactiveSessionStore sessionStore; + @Mock + private ReactiveAttachmentStore attachmentStore; + @Mock private RequestHeader.PrincipalHeader header; + @TempDir + Path storageRoot; + + private AttachmentFileCleaner fileCleaner; + private SessionServiceImpl service; @BeforeEach void setUp() { - service = new SessionServiceImpl(sessionStore); + AgenticProperties properties = new AgenticProperties(); + properties.setAttachmentStoragePath(storageRoot.toString()); + fileCleaner = new AttachmentFileCleaner(properties); + service = new SessionServiceImpl(sessionStore, attachmentStore, fileCleaner); } @Test @@ -69,4 +94,41 @@ class SessionServiceImplTest { .expectNext(page) .verifyComplete(); } + + @Test + void deleteUnlinksSoftDeletedAttachmentFilesAfterTheStoreDeleteCommits() throws Exception { + Path file = Files.writeString(storageRoot.resolve("tenant_1_user_2_conversation_conv-1-a.txt"), "payload"); + when(sessionStore.delete("conv", header)).thenReturn(Mono.just(1L)); + when(attachmentStore.listDeletedFilePaths("conv", header)).thenReturn(Flux.just(file.toString(), " ")); + + StepVerifier.create(service.delete("conv", header)).expectNext(1L).verifyComplete(); + + assertThat(Files.exists(file)) + .as("the soft-deleted attachment's file must be unlinked") + .isFalse(); + verify(attachmentStore).listDeletedFilePaths("conv", header); + } + + @Test + void deleteKeepsTheCommittedResultWhenAFileIsOutsideTheStorageRoot() throws Exception { + Path outside = Files.createTempDirectory("dc3-session-outside"); + Path escaped = Files.writeString(outside.resolve("escaped.txt"), "payload"); + when(sessionStore.delete("conv", header)).thenReturn(Mono.just(1L)); + when(attachmentStore.listDeletedFilePaths("conv", header)).thenReturn(Flux.just(escaped.toString())); + + StepVerifier.create(service.delete("conv", header)).expectNext(1L).verifyComplete(); + + assertThat(Files.exists(escaped)) + .as("the boundary guard must refuse out-of-root unlinks") + .isTrue(); + } + + @Test + void deleteSkipsFileCleanupWhenTheSessionWasAlreadyDeleted() { + when(sessionStore.delete("conv", header)).thenReturn(Mono.just(0L)); + + StepVerifier.create(service.delete("conv", header)).expectNext(0L).verifyComplete(); + + verify(attachmentStore, never()).listDeletedFilePaths(any(), any()); + } } diff --git a/dc3-db/dc3-db-agentic/pom.xml b/dc3-db/dc3-db-agentic/pom.xml index 734977763..0378f10aa 100644 --- a/dc3-db/dc3-db-agentic/pom.xml +++ b/dc3-db/dc3-db-agentic/pom.xml @@ -45,17 +45,12 @@ dc3-common-agentic - + org.springframework.boot spring-boot-starter-test test - - org.junit.jupiter - junit-jupiter - test - io.projectreactor reactor-test diff --git a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcActionStore.java b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcActionStore.java index 8890f3eaf..c8d7aeb0b 100644 --- a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcActionStore.java +++ b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcActionStore.java @@ -38,7 +38,14 @@ import reactor.core.publisher.Mono; import tools.jackson.core.type.TypeReference; import tools.jackson.databind.ObjectMapper; -/** Explicit SQL adapter for agentic actions. */ +/** + * Explicit SQL adapter for agentic actions. + * + *

Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through + * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone. + * The {@link ActionBO} wall-clock fields stay UTC-naive; reads normalize through + * {@link DatabaseInstant#toLocalDateTimeUtc}. + */ @Repository @ConditionalOnClass({DatabaseClient.class, PageTransaction.class}) @RequiredArgsConstructor @@ -69,6 +76,7 @@ public class R2dbcActionStore implements ReactiveActionStore { if (action.getId() == null) { action.setId(UuidV7.nextLong()); } + Instant now = Instant.now(); String sql = "INSERT INTO " + TABLE + " (id, action_id, conversation_id, action_type, title, description, payload, status, expire_time, " + "tenant_id, user_id, remark, creator_id, creator_name, create_time, operator_id, operator_name, operate_time, deleted) " @@ -91,14 +99,15 @@ public class R2dbcActionStore implements ReactiveActionStore { .bind("remark", valueOrEmpty(action.getRemark())) .bind("creator_id", valueOrZero(action.getCreatorId(), action.getUserId())) .bind("creator_name", valueOrEmpty(action.getCreatorName())) - .bind("create_time", utc(action.getCreateTime(), Instant.now())) + .bind("create_time", dialect.bindInstant(instant(action.getCreateTime(), now))) .bind("operator_id", valueOrZero(action.getOperatorId(), action.getUserId())) .bind("operator_name", valueOrEmpty(action.getOperatorName())) - .bind("operate_time", utc(action.getOperateTime(), Instant.now())); + .bind("operate_time", dialect.bindInstant(instant(action.getOperateTime(), now))); if (action.getExpireTime() == null) { - statement = statement.bindNull("expire_time", LocalDateTime.class); + statement = statement.bindNull("expire_time", Instant.class); } else { - statement = statement.bind("expire_time", action.getExpireTime()); + statement = statement.bind( + "expire_time", dialect.bindInstant(action.getExpireTime().toInstant(ZoneOffset.UTC))); } return statement .fetch() @@ -133,7 +142,7 @@ public class R2dbcActionStore implements ReactiveActionStore { return Mono.just(OffsetPage.of(java.util.List.of(), offset, limit, 0)); } new PageRequest(offset, limit); - LocalDateTime current = utc(now, Instant.now()); + Instant current = instant(now, Instant.now()); String predicate = " FROM " + TABLE + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id AND user_id = :user_id" + " AND status = :status AND deleted = 0" @@ -145,7 +154,7 @@ public class R2dbcActionStore implements ReactiveActionStore { .bind("tenant_id", header.getTenantId()) .bind("user_id", header.getUserId()) .bind("status", AgenticActionStatusEnum.PENDING.getIndex()) - .bind("now", current) + .bind("now", dialect.bindInstant(current)) .bind("limit", limit) .bind("offset", offset); DatabaseClient.GenericExecuteSpec countStatement = databaseClient @@ -154,7 +163,7 @@ public class R2dbcActionStore implements ReactiveActionStore { .bind("tenant_id", header.getTenantId()) .bind("user_id", header.getUserId()) .bind("status", AgenticActionStatusEnum.PENDING.getIndex()) - .bind("now", current); + .bind("now", dialect.bindInstant(current)); Mono total = countStatement .map((row, metadata) -> { Number value = row.get("total", Number.class); @@ -174,7 +183,7 @@ public class R2dbcActionStore implements ReactiveActionStore { if (nextStatus == null || nextStatus == AgenticActionStatusEnum.PENDING) { return Mono.error(new IllegalArgumentException("nextStatus must be a terminal action status")); } - LocalDateTime current = utc(now, Instant.now()); + Instant current = instant(now, Instant.now()); return databaseClient .sql("UPDATE " + TABLE + " SET status = :next_status, operator_id = :operator_id, " + "operator_name = :operator_name, operate_time = :operate_time" @@ -184,12 +193,12 @@ public class R2dbcActionStore implements ReactiveActionStore { .bind("next_status", nextStatus.getIndex()) .bind("operator_id", header.getUserId()) .bind("operator_name", valueOrEmpty(header.getUserName())) - .bind("operate_time", current) + .bind("operate_time", dialect.bindInstant(current)) .bind("action_id", actionId) .bind("tenant_id", header.getTenantId()) .bind("user_id", header.getUserId()) .bind("pending_status", AgenticActionStatusEnum.PENDING.getIndex()) - .bind("now", current) + .bind("now", dialect.bindInstant(current)) .fetch() .rowsUpdated() .flatMap(rows -> rows == 1 ? get(actionId, header) : Mono.empty()); @@ -215,7 +224,7 @@ public class R2dbcActionStore implements ReactiveActionStore { .bind("remark", valueOrEmpty(remark)) .bind("operator_id", header.getUserId()) .bind("operator_name", valueOrEmpty(header.getUserName())) - .bind("operate_time", utc(now, Instant.now())) + .bind("operate_time", dialect.bindInstant(instant(now, Instant.now()))) .bind("action_id", actionId) .bind("tenant_id", header.getTenantId()) .bind("user_id", header.getUserId()) @@ -292,11 +301,12 @@ public class R2dbcActionStore implements ReactiveActionStore { return value == null ? "" : value; } - private LocalDateTime utc(LocalDateTime value, Instant fallback) { - return value == null ? utc(fallback, Instant.now()) : value; + /** Interpret a BO wall-clock value as UTC and fall back to the given instant when absent. */ + private Instant instant(LocalDateTime value, Instant fallback) { + return value == null ? fallback : value.toInstant(ZoneOffset.UTC); } - private LocalDateTime utc(Instant value, Instant fallback) { - return LocalDateTime.ofInstant(value == null ? fallback : value, ZoneOffset.UTC); + private Instant instant(Instant value, Instant fallback) { + return value == null ? fallback : value; } } diff --git a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcAttachmentStore.java b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcAttachmentStore.java index d5d68d144..bc2822e35 100644 --- a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcAttachmentStore.java +++ b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcAttachmentStore.java @@ -19,10 +19,9 @@ package io.github.pnoker.db.agentic; import io.github.pnoker.common.agentic.entity.bo.AttachmentBO; import io.github.pnoker.common.agentic.repository.ReactiveAttachmentStore; import io.github.pnoker.common.entity.common.RequestHeader; +import io.github.pnoker.db.core.dialect.R2dbcDialect; import io.github.pnoker.db.core.time.DatabaseInstant; import java.time.Instant; -import java.time.LocalDateTime; -import java.time.ZoneOffset; import java.util.Collection; import java.util.Objects; import java.util.stream.Collectors; @@ -34,7 +33,14 @@ import org.springframework.transaction.reactive.TransactionalOperator; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -/** Explicit SQL adapter for attachment metadata. */ +/** + * Explicit SQL adapter for attachment metadata. + * + *

Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through + * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone. + * The {@link AttachmentBO} wall-clock fields stay UTC-naive; reads normalize through + * {@link DatabaseInstant#toLocalDateTimeUtc}. + */ @Repository @ConditionalOnClass({DatabaseClient.class, TransactionalOperator.class}) @RequiredArgsConstructor @@ -47,18 +53,34 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore { private final DatabaseClient databaseClient; private final TransactionalOperator transactionalOperator; + private final R2dbcDialect dialect; @Override public Mono save(AttachmentBO attachment) { Objects.requireNonNull(attachment, "attachment must not be null"); + Instant now = Instant.now(); String sql = "INSERT INTO " + TABLE + " (conversation_id, file_name, content_type, size, file_path, tenant_id, user_id," + " creator_id, creator_name, operator_id, operator_name, create_time, operate_time, deleted)" + " VALUES (:conversation_id, :file_name, :content_type, :size, :file_path, :tenant_id, :user_id," + " :creator_id, :creator_name, :operator_id, :operator_name, :create_time, :operate_time, 0)"; - LocalDateTime now = utcNow(); + // Guard on the live session row under the same lock the session delete takes: a save + // racing a delete either lands before the delete soft-deletes the children or fails + // here, never surviving as a live row of a soft-deleted conversation. + Mono sessionGuard = databaseClient + .sql("SELECT id FROM dc3_agentic.dc3_session" + + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id" + + " AND user_id = :user_id AND deleted = 0 FOR UPDATE") + .bind("conversation_id", attachment.getConversationId()) + .bind("tenant_id", attachment.getTenantId()) + .bind("user_id", attachment.getUserId()) + .fetch() + .all() + .switchIfEmpty(Mono.error( + new IllegalStateException("session must exist and must be live before saving an attachment"))) + .then(); return transactionalOperator - .transactional(databaseClient + .transactional(sessionGuard.then(databaseClient .sql(sql) .bind("conversation_id", attachment.getConversationId()) .bind("file_name", attachment.getFileName()) @@ -71,14 +93,14 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore { .bind("creator_name", attachment.getCreatorName()) .bind("operator_id", attachment.getOperatorId()) .bind("operator_name", attachment.getOperatorName()) - .bind("create_time", now) - .bind("operate_time", now) + .bind("create_time", dialect.bindInstant(now)) + .bind("operate_time", dialect.bindInstant(now)) .fetch() .rowsUpdated() .flatMap(rows -> rows == 1 ? getByPath(attachment.getFilePath(), attachment.getTenantId(), attachment.getUserId()) : Mono.error( - new IllegalStateException("attachment insert affected " + rows + " rows")))) + new IllegalStateException("attachment insert affected " + rows + " rows"))))) .switchIfEmpty(Mono.error(new IllegalStateException("attachment insert returned no row"))); } @@ -97,6 +119,21 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore { .all(); } + @Override + public Flux listDeletedFilePaths(String conversationId, RequestHeader.PrincipalHeader header) { + if (header == null) return Flux.error(new IllegalArgumentException("header must not be null")); + if (conversationId == null || conversationId.isBlank()) return Flux.empty(); + return databaseClient + .sql("SELECT file_path FROM " + TABLE + + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id" + + " AND user_id = :user_id AND deleted = 1") + .bind("conversation_id", conversationId) + .bind("tenant_id", header.getTenantId()) + .bind("user_id", header.getUserId()) + .map((row, metadata) -> row.get("file_path", String.class)) + .all(); + } + @Override public Flux getByIds(Collection ids, RequestHeader.PrincipalHeader header) { if (header == null) return Flux.error(new IllegalArgumentException("header must not be null")); @@ -139,7 +176,7 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore { + " operate_time = :operate_time WHERE id = :id AND tenant_id = :tenant_id" + " AND user_id = :user_id AND deleted = 0") .bind("operator_id", header.getUserId()) - .bind("operate_time", utcNow()) + .bind("operate_time", dialect.bindInstant(Instant.now())) .bind("id", id) .bind("tenant_id", header.getTenantId()) .bind("user_id", header.getUserId()) @@ -179,8 +216,4 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore { attachment.setOperateTime(DatabaseInstant.toLocalDateTimeUtc(row.get("operate_time"))); return attachment; } - - private LocalDateTime utcNow() { - return LocalDateTime.ofInstant(Instant.now(), ZoneOffset.UTC); - } } diff --git a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcMessageStore.java b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcMessageStore.java index 6abc8e007..e5bb492c9 100644 --- a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcMessageStore.java +++ b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcMessageStore.java @@ -24,8 +24,6 @@ import io.github.pnoker.common.enums.AgenticMessageStatusEnum; import io.github.pnoker.db.core.dialect.R2dbcDialect; import io.github.pnoker.db.core.time.DatabaseInstant; import java.time.Instant; -import java.time.LocalDateTime; -import java.time.ZoneOffset; import lombok.RequiredArgsConstructor; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.r2dbc.core.DatabaseClient; @@ -35,7 +33,14 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import tools.jackson.databind.ObjectMapper; -/** Explicit SQL adapter for the agentic message store. */ +/** + * Explicit SQL adapter for the agentic message store. + * + *

Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through + * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone. + * The {@link MessageBO} wall-clock fields stay UTC-naive; reads normalize through + * {@link DatabaseInstant#toLocalDateTimeUtc}. + */ @Repository @ConditionalOnClass({DatabaseClient.class, TransactionalOperator.class}) @RequiredArgsConstructor @@ -131,8 +136,8 @@ public class R2dbcMessageStore implements ReactiveMessageStore { .bind("creator_name", header.getUserName() == null ? "" : header.getUserName()) .bind("operator_id", header.getUserId()) .bind("operator_name", header.getUserName() == null ? "" : header.getUserName()) - .bind("create_time", LocalDateTime.ofInstant(now, ZoneOffset.UTC)) - .bind("operate_time", LocalDateTime.ofInstant(now, ZoneOffset.UTC)) + .bind("create_time", dialect.bindInstant(now)) + .bind("operate_time", dialect.bindInstant(now)) .fetch() .rowsUpdated() .flatMap(rows -> rows == 1 @@ -202,7 +207,7 @@ public class R2dbcMessageStore implements ReactiveMessageStore { .sql("UPDATE " + TABLE + " SET deleted = 1, operate_time = :operate_time" + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id" + " AND user_id = :user_id AND deleted = 0") - .bind("operate_time", LocalDateTime.ofInstant(Instant.now(), ZoneOffset.UTC)) + .bind("operate_time", dialect.bindInstant(Instant.now())) .bind("conversation_id", conversationId) .bind("tenant_id", header.getTenantId()) .bind("user_id", header.getUserId()) diff --git a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcModelConfigStore.java b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcModelConfigStore.java index 6b9e491b0..d5669fc7e 100644 --- a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcModelConfigStore.java +++ b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcModelConfigStore.java @@ -24,7 +24,9 @@ import io.github.pnoker.common.enums.ConnectivityStatusEnum; import io.github.pnoker.common.enums.DefaultFlagEnum; import io.github.pnoker.common.enums.EnableFlagEnum; import io.github.pnoker.common.utils.UuidV7; +import io.github.pnoker.db.core.dialect.R2dbcDialect; import io.github.pnoker.db.core.time.DatabaseInstant; +import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneOffset; import lombok.RequiredArgsConstructor; @@ -35,7 +37,14 @@ import org.springframework.transaction.reactive.TransactionalOperator; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -/** Explicit SQL adapter for agentic model configurations. */ +/** + * Explicit SQL adapter for agentic model configurations. + * + *

Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through + * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone. + * The {@link ModelConfigBO} wall-clock fields stay UTC-naive; reads normalize through + * {@link DatabaseInstant#toLocalDateTimeUtc}. + */ @Repository @ConditionalOnClass({DatabaseClient.class, TransactionalOperator.class}) @RequiredArgsConstructor @@ -50,6 +59,7 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore { private final DatabaseClient databaseClient; private final TransactionalOperator transactionalOperator; + private final R2dbcDialect dialect; @Override public Flux list(RequestHeader.PrincipalHeader header, boolean enabledOnly) { @@ -142,10 +152,10 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore { .bind("remark", value.getRemark()) .bind("creator_id", value.getCreatorId()) .bind("creator_name", value.getCreatorName()) - .bind("create_time", utc(value.getCreateTime())) + .bind("create_time", dialect.bindInstant(instant(value.getCreateTime()))) .bind("operator_id", value.getOperatorId()) .bind("operator_name", value.getOperatorName()) - .bind("operate_time", utc(value.getOperateTime())) + .bind("operate_time", dialect.bindInstant(instant(value.getOperateTime()))) .fetch() .rowsUpdated() .flatMap(rows -> rows == 1 @@ -185,7 +195,7 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore { .bind("remark", value.getRemark()) .bind("operator_id", value.getOperatorId()) .bind("operator_name", value.getOperatorName()) - .bind("operate_time", utc(value.getOperateTime())) + .bind("operate_time", dialect.bindInstant(instant(value.getOperateTime()))) .bind("id", value.getId()) .bind("tenant_id", value.getTenantId()) .fetch() @@ -202,7 +212,7 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore { .sql("UPDATE " + TABLE + " SET deleted = 1, operator_id = :operator_id," + " operate_time = :operate_time WHERE id = :id AND tenant_id = :tenant_id AND deleted = 0") .bind("operator_id", header.getUserId()) - .bind("operate_time", utcNow()) + .bind("operate_time", dialect.bindInstant(Instant.now())) .bind("id", id) .bind("tenant_id", header.getTenantId()) .fetch() @@ -228,8 +238,9 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore { ? spec.bindNull("last_check_status", Byte.class) : spec.bind("last_check_status", profile.status().getIndex()); spec = profile.time() == null - ? spec.bindNull("last_check_time", LocalDateTime.class) - : spec.bind("last_check_time", profile.time()); + ? spec.bindNull("last_check_time", Instant.class) + : spec.bind( + "last_check_time", dialect.bindInstant(profile.time().toInstant(ZoneOffset.UTC))); spec = profile.latencyMs() == null ? spec.bindNull("last_check_latency_ms", Integer.class) : spec.bind("last_check_latency_ms", profile.latencyMs()); @@ -257,7 +268,7 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore { return databaseClient .sql("UPDATE " + TABLE + " SET default_flag = 0, operate_time = :operate_time" + " WHERE tenant_id = :tenant_id AND default_flag = 1 AND deleted = 0 AND id <> :id") - .bind("operate_time", utcNow()) + .bind("operate_time", dialect.bindInstant(Instant.now())) .bind("tenant_id", tenantId) .bind("id", exceptId) .fetch() @@ -332,12 +343,9 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore { : value.getEnableFlag().getIndex(); } - private LocalDateTime utc(LocalDateTime value) { - return value == null ? utcNow() : value; - } - - private LocalDateTime utcNow() { - return LocalDateTime.now(ZoneOffset.UTC); + /** Interpret a BO wall-clock value as UTC, defaulting to now when absent. */ + private Instant instant(LocalDateTime value) { + return value == null ? Instant.now() : value.toInstant(ZoneOffset.UTC); } private LocalDateTime time(Object value) { diff --git a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcSessionStore.java b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcSessionStore.java index 11f2ce97e..6b6c359c8 100644 --- a/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcSessionStore.java +++ b/dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcSessionStore.java @@ -19,7 +19,6 @@ package io.github.pnoker.db.agentic; import io.github.pnoker.common.agentic.entity.bo.SessionBO; import io.github.pnoker.common.agentic.entity.model.SessionExt; import io.github.pnoker.common.agentic.repository.ReactiveSessionStore; -import io.github.pnoker.common.agentic.service.impl.AttachmentFileCleaner; import io.github.pnoker.common.constant.service.AgenticConstant; import io.github.pnoker.common.entity.common.RequestHeader; import io.github.pnoker.db.core.dialect.R2dbcDialect; @@ -28,22 +27,23 @@ import io.github.pnoker.db.core.page.SortSpec; import io.github.pnoker.db.core.time.DatabaseInstant; import io.github.pnoker.db.core.transaction.PageTransaction; import java.time.Instant; -import java.time.LocalDateTime; -import java.time.ZoneOffset; -import java.util.List; import lombok.RequiredArgsConstructor; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.dao.DataIntegrityViolationException; import org.springframework.r2dbc.core.DatabaseClient; import org.springframework.stereotype.Repository; import org.springframework.transaction.reactive.TransactionalOperator; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.util.function.Tuple2; -import reactor.util.function.Tuples; import tools.jackson.databind.ObjectMapper; -/** Explicit SQL adapter for agentic sessions. */ +/** + * Explicit SQL adapter for agentic sessions. + * + *

Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through + * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone. + * The {@link SessionBO} wall-clock fields stay UTC-naive; reads normalize through + * {@link DatabaseInstant#toLocalDateTimeUtc}. + */ @Repository @ConditionalOnClass({DatabaseClient.class, TransactionalOperator.class, PageTransaction.class}) @RequiredArgsConstructor @@ -58,22 +58,39 @@ public class R2dbcSessionStore implements ReactiveSessionStore { private final PageTransaction pageTransaction; private final ObjectMapper objectMapper; private final R2dbcDialect dialect; - private final AttachmentFileCleaner attachmentFileCleaner; @Override public Mono touch(String conversationId, SessionExt sessionExt, RequestHeader.PrincipalHeader header) { + return touchAttempt(conversationId, sessionExt, header, true); + } + + /** + * One touch attempt: lock the conversation's row regardless of tombstone state, then update, + * revive, or insert. A unique-key violation means a concurrent touch inserted the same + * conversation first, so the caller retries the whole decision once instead of surfacing the + * raw constraint error. + */ + private Mono touchAttempt( + String conversationId, + SessionExt sessionExt, + RequestHeader.PrincipalHeader header, + boolean retryOnConflict) { if (conversationId == null || conversationId.isBlank()) { return Mono.error(new IllegalArgumentException("conversationId must not be blank")); } if (header == null) return Mono.error(new IllegalArgumentException("header must not be null")); - Mono transaction = lock(conversationId, header) - .flatMap(existing -> updateExisting(conversationId, sessionExt, existing, header)) - .switchIfEmpty(insert(conversationId, serialize(sessionExt), header)); + Mono transaction = lockAny(conversationId, header) + .flatMap(locked -> locked.deleted() + ? revive(conversationId, sessionExt, locked.session(), header) + : updateRow(conversationId, sessionExt, null, locked.session(), header)) + .switchIfEmpty(Mono.defer(() -> insert(conversationId, serialize(sessionExt), header))); return transactionalOperator .transactional(transaction) .onErrorResume( DataIntegrityViolationException.class, - ignored -> get(conversationId, header).switchIfEmpty(Mono.error(ignored))); + ignored -> retryOnConflict + ? touchAttempt(conversationId, sessionExt, header, false) + : Mono.error(ignored)); } @Override @@ -167,25 +184,11 @@ public class R2dbcSessionStore implements ReactiveSessionStore { return Mono.error(new IllegalArgumentException("conversationId must not be blank")); } if (header == null) return Mono.error(new IllegalArgumentException("header must not be null")); - return get(conversationId, header).flatMap(existing -> { - String nextTitle = title == null || title.isBlank() ? existing.getTitle() : title.trim(); - String nextExt = serialize(merge(existing.getSessionExt(), sessionExt)); - return databaseClient - .sql("UPDATE " + TABLE + " SET title = :title, session_ext = " - + dialect.jsonWriteExpression(":session_ext") + "," - + " operator_id = :operator_id, operate_time = :operate_time WHERE id = :id" - + " AND tenant_id = :tenant_id AND user_id = :user_id AND deleted = 0") - .bind("title", nextTitle) - .bind("session_ext", nextExt) - .bind("operator_id", header.getUserId()) - .bind("operate_time", utcNow()) - .bind("id", existing.getId()) - .bind("tenant_id", header.getTenantId()) - .bind("user_id", header.getUserId()) - .fetch() - .rowsUpdated() - .flatMap(rows -> rows == 1 ? get(conversationId, header) : Mono.empty()); - }); + // The row lock serializes concurrent updates and the merge reads the locked row, so + // concurrent preference changes merge instead of the last writer silently winning. + Mono transaction = lock(conversationId, header) + .flatMap(existing -> updateRow(conversationId, sessionExt, title, existing, header)); + return transactionalOperator.transactional(transaction); } @Override @@ -194,62 +197,44 @@ public class R2dbcSessionStore implements ReactiveSessionStore { return Mono.error(new IllegalArgumentException("conversationId must not be blank")); } if (header == null) return Mono.error(new IllegalArgumentException("header must not be null")); - Mono deleteMessages = databaseClient - .sql("UPDATE dc3_agentic.dc3_message SET deleted = 1," - + " operate_time = :operate_time WHERE conversation_id = :conversation_id" - + " AND tenant_id = :tenant_id AND user_id = :user_id AND deleted = 0") - .bind("operate_time", utcNow()) + // Lock the live session row first: attachment saves take the same lock before inserting, + // so a concurrent upload either lands before this delete (and is soft-deleted with the + // rest) or fails its guard once the session is gone. Children then the session row. + Mono transaction = lockSessionId(conversationId, header) + .flatMap(sessionId -> softDeleteChildren("dc3_agentic.dc3_message", conversationId, header) + .then(softDeleteChildren("dc3_agentic.dc3_attachment", conversationId, header)) + .then(softDeleteChildren("dc3_agentic.dc3_action", conversationId, header)) + .then(deleteSession(conversationId, header))) + .defaultIfEmpty(0L); + return transactionalOperator.transactional(transaction); + } + + /** Lock the live session row and emit its id, or empty when the session is already deleted. */ + private Mono lockSessionId(String conversationId, RequestHeader.PrincipalHeader header) { + return databaseClient + .sql("SELECT id FROM " + TABLE + + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id" + + " AND user_id = :user_id AND deleted = 0 FOR UPDATE") .bind("conversation_id", conversationId) .bind("tenant_id", header.getTenantId()) .bind("user_id", header.getUserId()) - .fetch() - .rowsUpdated() - .map(Long::valueOf); - Mono deleteAttachments = softDeleteChildren("dc3_agentic.dc3_attachment", conversationId, header); - Mono deleteActions = softDeleteChildren("dc3_agentic.dc3_action", conversationId, header); - Mono deleteSession = databaseClient + .map((row, metadata) -> row.get("id", Long.class)) + .one(); + } + + private Mono deleteSession(String conversationId, RequestHeader.PrincipalHeader header) { + return databaseClient .sql("UPDATE " + TABLE + " SET deleted = 1," + " operator_id = :operator_id, operate_time = :operate_time WHERE conversation_id = :conversation_id" + " AND tenant_id = :tenant_id AND user_id = :user_id AND deleted = 0") .bind("operator_id", header.getUserId()) - .bind("operate_time", utcNow()) + .bind("operate_time", dialect.bindInstant(Instant.now())) .bind("conversation_id", conversationId) .bind("tenant_id", header.getTenantId()) .bind("user_id", header.getUserId()) .fetch() .rowsUpdated() .map(Long::valueOf); - return transactionalOperator - .transactional(listAttachmentFilePaths(conversationId, header) - .collectList() - .flatMap(filePaths -> deleteMessages - .then(deleteAttachments) - .then(deleteActions) - .then(deleteSession) - .map(deleted -> Tuples.of(deleted, filePaths)))) - .flatMap(deletedSession -> unlinkAttachmentFiles(deletedSession).thenReturn(deletedSession.getT1())); - } - - /** Read attachment file locations while the rows are still live, so the files can be unlinked after commit. */ - private Flux listAttachmentFilePaths(String conversationId, RequestHeader.PrincipalHeader header) { - return databaseClient - .sql("SELECT file_path FROM dc3_agentic.dc3_attachment" - + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id" - + " AND user_id = :user_id AND deleted = 0") - .bind("conversation_id", conversationId) - .bind("tenant_id", header.getTenantId()) - .bind("user_id", header.getUserId()) - .map((row, metadata) -> row.get("file_path", String.class)) - .all(); - } - - /** - * Unlink the conversation's stored attachment files after the session delete commits. Runs - * strictly after the transaction so file cleanup can never roll the delete back, and is - * warn-only so a missing or undeletable file never fails the committed delete. - */ - private Mono unlinkAttachmentFiles(Tuple2> deletedSession) { - return attachmentFileCleaner.unlinkAll(deletedSession.getT2()); } private Mono softDeleteChildren(String table, String conversationId, RequestHeader.PrincipalHeader header) { @@ -257,7 +242,7 @@ public class R2dbcSessionStore implements ReactiveSessionStore { .sql("UPDATE " + table + " SET deleted = 1, operate_time = :operate_time" + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id" + " AND user_id = :user_id AND deleted = 0") - .bind("operate_time", utcNow()) + .bind("operate_time", dialect.bindInstant(Instant.now())) .bind("conversation_id", conversationId) .bind("tenant_id", header.getTenantId()) .bind("user_id", header.getUserId()) @@ -266,6 +251,7 @@ public class R2dbcSessionStore implements ReactiveSessionStore { .map(Long::valueOf); } + /** Lock the live session row for update; empty when the conversation has no live row. */ private Mono lock(String conversationId, RequestHeader.PrincipalHeader header) { return databaseClient .sql("SELECT " + COLUMNS + " FROM " + TABLE @@ -278,16 +264,75 @@ public class R2dbcSessionStore implements ReactiveSessionStore { .one(); } - private Mono updateExisting( - String conversationId, SessionExt sessionExt, SessionBO existing, RequestHeader.PrincipalHeader header) { - String nextJson = serialize(merge(existing.getSessionExt(), sessionExt)); + /** + * Lock the conversation's row regardless of soft-delete state. The natural key is unique + * across live and deleted rows (it is the foreign-key target for the child tables), so the + * tombstone row is the row that must be locked and revived to reuse the conversation id. + */ + private Mono lockAny(String conversationId, RequestHeader.PrincipalHeader header) { return databaseClient - .sql("UPDATE " + TABLE + " SET session_ext = " + dialect.jsonWriteExpression(":session_ext") - + ", operator_id = :operator_id," + " operate_time = :operate_time WHERE id = :id") - .bind("session_ext", nextJson) + .sql("SELECT " + COLUMNS + " FROM " + TABLE + + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id" + + " AND user_id = :user_id LIMIT 1 FOR UPDATE") + .bind("conversation_id", conversationId) + .bind("tenant_id", header.getTenantId()) + .bind("user_id", header.getUserId()) + .map((row, metadata) -> new LockedSession( + map(row, metadata), row.get("deleted", Number.class).intValue() != 0)) + .one(); + } + + /** + * Revive a soft-deleted session row as a brand-new conversation: the tombstone's title, + * extension, creator, and creation time are reset, because the deleted conversation's + * messages and attachments stay soft-deleted and must never reappear. + */ + private Mono revive( + String conversationId, SessionExt sessionExt, SessionBO tombstone, RequestHeader.PrincipalHeader header) { + return databaseClient + .sql("UPDATE " + TABLE + " SET deleted = 0, title = :title, session_ext = " + + dialect.jsonWriteExpression(":session_ext") + ", creator_id = :creator_id," + + " creator_name = :creator_name, create_time = :create_time, operator_id = :operator_id," + + " operate_time = :operate_time WHERE id = :id AND tenant_id = :tenant_id" + + " AND user_id = :user_id") + .bind("title", AgenticConstant.Session.DEFAULT_TITLE) + .bind("session_ext", serialize(sessionExt)) + .bind("creator_id", header.getUserId()) + .bind("creator_name", header.getUserName() == null ? "" : header.getUserName()) + .bind("create_time", dialect.bindInstant(Instant.now())) .bind("operator_id", header.getUserId()) - .bind("operate_time", utcNow()) + .bind("operate_time", dialect.bindInstant(Instant.now())) + .bind("id", tombstone.getId()) + .bind("tenant_id", header.getTenantId()) + .bind("user_id", header.getUserId()) + .fetch() + .rowsUpdated() + .flatMap(rows -> rows == 1 + ? get(conversationId, header) + : Mono.error(new IllegalStateException("session revive affected " + rows + " rows"))); + } + + /** Merge the incoming extension onto the locked row and persist it with the optional title. */ + private Mono updateRow( + String conversationId, + SessionExt sessionExt, + String title, + SessionBO existing, + RequestHeader.PrincipalHeader header) { + String nextTitle = title == null || title.isBlank() ? existing.getTitle() : title.trim(); + String nextExt = serialize(merge(existing.getSessionExt(), sessionExt)); + return databaseClient + .sql("UPDATE " + TABLE + " SET title = :title, session_ext = " + + dialect.jsonWriteExpression(":session_ext") + "," + + " operator_id = :operator_id, operate_time = :operate_time WHERE id = :id" + + " AND tenant_id = :tenant_id AND user_id = :user_id AND deleted = 0") + .bind("title", nextTitle) + .bind("session_ext", nextExt) + .bind("operator_id", header.getUserId()) + .bind("operate_time", dialect.bindInstant(Instant.now())) .bind("id", existing.getId()) + .bind("tenant_id", header.getTenantId()) + .bind("user_id", header.getUserId()) .fetch() .rowsUpdated() .flatMap(rows -> rows == 1 @@ -308,8 +353,8 @@ public class R2dbcSessionStore implements ReactiveSessionStore { .bind("user_id", header.getUserId()) .bind("creator_id", header.getUserId()) .bind("operator_id", header.getUserId()) - .bind("create_time", utcNow()) - .bind("operate_time", utcNow()) + .bind("create_time", dialect.bindInstant(Instant.now())) + .bind("operate_time", dialect.bindInstant(Instant.now())) .fetch() .rowsUpdated() .flatMap(rows -> rows == 1 @@ -376,7 +421,6 @@ public class R2dbcSessionStore implements ReactiveSessionStore { return merged; } - private LocalDateTime utcNow() { - return LocalDateTime.ofInstant(Instant.now(), ZoneOffset.UTC); - } + /** A session row read under lock together with its soft-delete tombstone flag. */ + private record LockedSession(SessionBO session, boolean deleted) {} } diff --git a/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/AgenticStoreLayeringTest.java b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/AgenticStoreLayeringTest.java new file mode 100644 index 000000000..bf6e8bcc3 --- /dev/null +++ b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/AgenticStoreLayeringTest.java @@ -0,0 +1,56 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package io.github.pnoker.db.agentic; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.stream.Stream; +import org.junit.jupiter.api.Test; + +/** + * Layering guard: the database adapters must stay pure SQL ports. They implement the + * {@code Reactive*Store} contracts of {@code dc3-common-agentic} and must never import or + * orchestrate service-layer implementation classes (business rules, filesystem cleanup), so the + * dependency direction never reverses. + */ +class AgenticStoreLayeringTest { + + @Test + void storeAdaptersNeverImportServiceImplementations() throws IOException { + List offenders; + try (Stream sources = Files.walk(Path.of("src", "main", "java"))) { + offenders = sources.filter(path -> path.toString().endsWith(".java")) + .filter(path -> containsServiceImport(path)) + .toList(); + } + assertThat(offenders) + .as("db adapters must not reference io.github.pnoker.common.agentic.service.impl") + .isEmpty(); + } + + private boolean containsServiceImport(Path path) { + try { + return Files.readString(path).contains("io.github.pnoker.common.agentic.service.impl"); + } catch (IOException exception) { + throw new IllegalStateException("cannot read " + path, exception); + } + } +} diff --git a/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcActionStoreTest.java b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcActionStoreTest.java new file mode 100644 index 000000000..b4a715902 --- /dev/null +++ b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcActionStoreTest.java @@ -0,0 +1,118 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package io.github.pnoker.db.agentic; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.github.pnoker.common.agentic.entity.bo.ActionBO; +import io.github.pnoker.common.entity.common.RequestHeader; +import io.github.pnoker.common.enums.AgenticActionStatusEnum; +import io.github.pnoker.db.core.dialect.StandardR2dbcDialect; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.test.StepVerifier; +import tools.jackson.databind.ObjectMapper; + +/** Regression coverage for timezone-safe timestamp binds across the action store statements. */ +class R2dbcActionStoreTest { + + private static final long TENANT_ID = 7L; + private static final long USER_ID = 42L; + private static final String CONVERSATION_ID = "conv-1"; + + private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase(); + private final R2dbcActionStore store = new R2dbcActionStore( + database, + ScriptedAgenticDatabase.passThroughPageTransaction(), + new ObjectMapper(), + new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint")); + + private RequestHeader.PrincipalHeader header; + + @BeforeEach + void setUp() { + header = new RequestHeader.PrincipalHeader(); + header.setTenantId(TENANT_ID); + header.setPrincipalId(USER_ID); + header.setPrincipalName("operator"); + } + + @Test + void createBindsEveryTimestampAsAnInstant() { + ActionBO action = action(LocalDateTime.now(ZoneOffset.UTC).plusMinutes(10)); + + StepVerifier.create(store.create(action)).expectNextCount(1).verifyComplete(); + + ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_action"); + assertThat(insert.binds().get("expire_time")) + .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock") + .isInstanceOf(Instant.class); + assertThat(insert.binds().get("create_time")).isInstanceOf(Instant.class); + assertThat(insert.binds().get("operate_time")).isInstanceOf(Instant.class); + } + + @Test + void createBindsANullExpireTimeWithTheInstantType() { + StepVerifier.create(store.create(action(null))).expectNextCount(1).verifyComplete(); + + ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_action"); + assertThat(insert.nulls()).containsEntry("expire_time", Instant.class); + assertThat(insert.binds()).doesNotContainKey("expire_time"); + } + + @Test + void listPendingBindsTheExpiryWindowAsInstants() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Live", "{}", false); + + StepVerifier.create(store.listPending(0, 10, CONVERSATION_ID, header, Instant.now())) + .expectNextCount(1) + .verifyComplete(); + + ScriptedAgenticDatabase.Execution query = database.firstMatching("FROM dc3_agentic.dc3_action"); + assertThat(query.binds().get("now")).isInstanceOf(Instant.class); + } + + @Test + void claimPendingBindsOperateTimeAsAnInstant() { + ActionBO created = action(null); + StepVerifier.create(store.create(created)).expectNextCount(1).verifyComplete(); + + StepVerifier.create(store.claimPending("action-1", header, AgenticActionStatusEnum.CONFIRMED, Instant.now())) + .expectNextCount(1) + .verifyComplete(); + + ScriptedAgenticDatabase.Execution update = database.firstMatching("UPDATE dc3_agentic.dc3_action"); + assertThat(update.binds().get("operate_time")).isInstanceOf(Instant.class); + assertThat(update.binds().get("now")).isInstanceOf(Instant.class); + } + + private ActionBO action(LocalDateTime expireTime) { + ActionBO action = new ActionBO(); + action.setActionId("action-1"); + action.setConversationId(CONVERSATION_ID); + action.setActionType("write"); + action.setTitle("Write value"); + action.setDescription("Writes a device point value"); + action.setTenantId(TENANT_ID); + action.setUserId(USER_ID); + action.setExpireTime(expireTime); + return action; + } +} diff --git a/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcAttachmentStoreTest.java b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcAttachmentStoreTest.java new file mode 100644 index 000000000..c82d19530 --- /dev/null +++ b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcAttachmentStoreTest.java @@ -0,0 +1,184 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package io.github.pnoker.db.agentic; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.github.pnoker.common.agentic.entity.bo.AttachmentBO; +import io.github.pnoker.common.entity.common.RequestHeader; +import io.github.pnoker.db.core.dialect.StandardR2dbcDialect; +import java.time.Instant; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.test.StepVerifier; + +/** + * Regression coverage for the attachment save guard: the insert runs in one transaction behind a + * live-session row lock, so a save racing a concurrent session delete either lands before the + * delete or fails with a clear error, never surviving as a zombie row of a soft-deleted + * conversation. Timestamps bind as {@link Instant} so TIMESTAMPTZ encoding cannot drift. + */ +class R2dbcAttachmentStoreTest { + + private static final long TENANT_ID = 7L; + private static final long USER_ID = 42L; + private static final String CONVERSATION_ID = "conv-1"; + + private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase(); + private final R2dbcAttachmentStore store = new R2dbcAttachmentStore( + database, + ScriptedAgenticDatabase.passThroughOperator(), + new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint")); + + private RequestHeader.PrincipalHeader header; + + @BeforeEach + void setUp() { + header = new RequestHeader.PrincipalHeader(); + header.setTenantId(TENANT_ID); + header.setPrincipalId(USER_ID); + header.setPrincipalName("operator"); + } + + @Test + void saveInsertsBehindTheLiveSessionGuardWhenTheSessionIsLive() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Live", "{}", false); + + StepVerifier.create(store.save(attachment("/store/note.txt"))) + .assertNext(saved -> { + assertThat(saved.getConversationId()).isEqualTo(CONVERSATION_ID); + assertThat(saved.getFilePath()).isEqualTo("/store/note.txt"); + }) + .verifyComplete(); + + int guardIndex = database.indexOfFirst("SELECT id FROM dc3_agentic.dc3_session"); + int insertIndex = database.indexOfFirst("INSERT INTO dc3_agentic.dc3_attachment"); + assertThat(guardIndex) + .as("the live-session guard must run before the insert in the same transaction") + .isGreaterThanOrEqualTo(0) + .isLessThan(insertIndex); + } + + @Test + void saveFailsClosedWhenTheSessionIsASoftDeletedTombstone() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Gone", "{}", true); + + StepVerifier.create(store.save(attachment("/store/note.txt"))) + .expectErrorSatisfies(error -> { + assertThat(error).isInstanceOf(IllegalStateException.class); + assertThat(error.getMessage()).contains("session must exist"); + }) + .verify(); + + assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_attachment")) + .as("a tombstoned session must never accept a new attachment row") + .isZero(); + } + + @Test + void saveFailsClosedWhenTheSessionDoesNotExist() { + StepVerifier.create(store.save(attachment("/store/note.txt"))) + .expectErrorSatisfies(error -> assertThat(error).isInstanceOf(IllegalStateException.class)) + .verify(); + + assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_attachment")) + .isZero(); + } + + @Test + void saveLosesTheRaceAgainstAConcurrentSessionDelete() throws Exception { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Doomed", "{}", false); + // Park the delete's first child soft-delete so the upload's guard subscribes while the + // delete transaction holds the session row lock. + ScriptedAgenticDatabase.Hold deleteHold = database.holdNext("UPDATE dc3_agentic.dc3_message"); + + R2dbcSessionStore sessionStore = sessionStore(); + CompletableFuture delete = + sessionStore.delete(CONVERSATION_ID, header).toFuture(); + deleteHold.awaitRouted(); + + CompletableFuture upload = + store.save(attachment("/store/late.txt")).toFuture(); + assertThat(upload.isDone()) + .as("the upload guard must wait on the session row lock held by the delete") + .isFalse(); + + deleteHold.release(); + assertThat(delete.get(5, TimeUnit.SECONDS)).isEqualTo(1L); + assertThat(upload) + .as("the upload must fail once the delete committed, instead of surviving it") + .isCompletedExceptionally(); + + assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_attachment")) + .as("no attachment row may outlive the session delete") + .isZero(); + assertThat(database.session(TENANT_ID, USER_ID, CONVERSATION_ID).deleted) + .isOne(); + } + + @Test + void listDeletedFilePathsStreamsOnlySoftDeletedRows() { + database.seedAttachment(TENANT_ID, USER_ID, CONVERSATION_ID, "/store/gone.txt", true); + database.seedAttachment(TENANT_ID, USER_ID, CONVERSATION_ID, "/store/live.txt", false); + + StepVerifier.create(store.listDeletedFilePaths(CONVERSATION_ID, header)) + .expectNext("/store/gone.txt") + .verifyComplete(); + } + + @Test + void saveBindsTimestampsAsInstantsSoEncodingIsTimezoneIndependent() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Live", "{}", false); + + StepVerifier.create(store.save(attachment("/store/note.txt"))) + .expectNextCount(1) + .verifyComplete(); + + ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_attachment"); + assertThat(insert.binds().get("create_time")) + .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock") + .isInstanceOf(Instant.class); + assertThat(insert.binds().get("operate_time")).isInstanceOf(Instant.class); + } + + private R2dbcSessionStore sessionStore() { + return new R2dbcSessionStore( + database, + ScriptedAgenticDatabase.passThroughOperator(), + org.mockito.Mockito.mock(io.github.pnoker.db.core.transaction.PageTransaction.class), + new tools.jackson.databind.ObjectMapper(), + new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint")); + } + + private AttachmentBO attachment(String path) { + AttachmentBO attachment = new AttachmentBO(); + attachment.setConversationId(CONVERSATION_ID); + attachment.setFileName("note.txt"); + attachment.setContentType("text/plain"); + attachment.setSize(7L); + attachment.setFilePath(path); + attachment.setTenantId(TENANT_ID); + attachment.setUserId(USER_ID); + attachment.setCreatorId(USER_ID); + attachment.setCreatorName("operator"); + attachment.setOperatorId(USER_ID); + attachment.setOperatorName("operator"); + return attachment; + } +} diff --git a/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcMessageStoreTest.java b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcMessageStoreTest.java new file mode 100644 index 000000000..1537407da --- /dev/null +++ b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcMessageStoreTest.java @@ -0,0 +1,100 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package io.github.pnoker.db.agentic; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.github.pnoker.common.agentic.entity.model.AgenticMessageContent; +import io.github.pnoker.common.entity.common.RequestHeader; +import io.github.pnoker.common.enums.AgenticMessageStatusEnum; +import io.github.pnoker.db.core.dialect.StandardR2dbcDialect; +import java.time.Instant; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.test.StepVerifier; +import tools.jackson.databind.ObjectMapper; + +/** Regression coverage for the message store save protocol and timezone-safe timestamp binds. */ +class R2dbcMessageStoreTest { + + private static final long TENANT_ID = 7L; + private static final long USER_ID = 42L; + private static final String CONVERSATION_ID = "conv-1"; + + private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase(); + private final R2dbcMessageStore store = new R2dbcMessageStore( + database, + ScriptedAgenticDatabase.passThroughOperator(), + new ObjectMapper(), + new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint")); + + private RequestHeader.PrincipalHeader header; + + @BeforeEach + void setUp() { + header = new RequestHeader.PrincipalHeader(); + header.setTenantId(TENANT_ID); + header.setPrincipalId(USER_ID); + header.setPrincipalName("operator"); + } + + @Test + void saveRunsTheSessionGuardBeforeInsertingAndBindsInstants() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Live", "{}", false); + + StepVerifier.create(store.save( + CONVERSATION_ID, + "user", + AgenticMessageContent.ofText("hello"), + "m1", + AgenticMessageStatusEnum.COMPLETED, + header)) + .assertNext(message -> { + assertThat(message.getConversationId()).isEqualTo(CONVERSATION_ID); + assertThat(message.getMessageIndex()).isZero(); + }) + .verifyComplete(); + + int guardIndex = database.indexOfFirst("SELECT id FROM dc3_agentic.dc3_session"); + int insertIndex = database.indexOfFirst("INSERT INTO dc3_agentic.dc3_message"); + assertThat(guardIndex).isGreaterThanOrEqualTo(0).isLessThan(insertIndex); + + ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_message"); + assertThat(insert.binds().get("create_time")) + .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock") + .isInstanceOf(Instant.class); + assertThat(insert.binds().get("operate_time")).isInstanceOf(Instant.class); + } + + @Test + void saveFailsClosedWhenTheSessionIsMissingOrDeleted() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Gone", "{}", true); + + StepVerifier.create(store.save( + CONVERSATION_ID, + "user", + AgenticMessageContent.ofText("hello"), + "m1", + AgenticMessageStatusEnum.COMPLETED, + header)) + .expectError(IllegalStateException.class) + .verify(); + + assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_message")) + .isZero(); + } +} diff --git a/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcModelConfigStoreTest.java b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcModelConfigStoreTest.java new file mode 100644 index 000000000..ac06f39c3 --- /dev/null +++ b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcModelConfigStoreTest.java @@ -0,0 +1,73 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package io.github.pnoker.db.agentic; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.github.pnoker.common.agentic.entity.model.ConnectivityCheckProfile; +import io.github.pnoker.common.entity.common.RequestHeader; +import io.github.pnoker.common.enums.ConnectivityStatusEnum; +import io.github.pnoker.db.core.dialect.StandardR2dbcDialect; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.test.StepVerifier; + +/** Regression coverage for timezone-safe timestamp binds across the model config statements. */ +class R2dbcModelConfigStoreTest { + + private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase(); + private final R2dbcModelConfigStore store = new R2dbcModelConfigStore( + database, + ScriptedAgenticDatabase.passThroughOperator(), + new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint")); + + private RequestHeader.PrincipalHeader header; + + @BeforeEach + void setUp() { + header = new RequestHeader.PrincipalHeader(); + header.setTenantId(7L); + header.setPrincipalId(42L); + header.setPrincipalName("operator"); + } + + @Test + void updateCheckProfileBindsTheCheckTimeAsAnInstant() { + ConnectivityCheckProfile profile = new ConnectivityCheckProfile( + ConnectivityStatusEnum.PASS, LocalDateTime.now(ZoneOffset.UTC), 120, null, null, null); + + StepVerifier.create(store.updateCheckProfile(1L, profile, header)) + .expectNext(true) + .verifyComplete(); + + ScriptedAgenticDatabase.Execution update = database.firstMatching("last_check_time = :last_check_time"); + assertThat(update.binds().get("last_check_time")) + .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock") + .isInstanceOf(Instant.class); + } + + @Test + void deleteBindsOperateTimeAsAnInstant() { + StepVerifier.create(store.delete(1L, header)).expectNext(true).verifyComplete(); + + ScriptedAgenticDatabase.Execution delete = database.firstMatching("SET deleted = 1"); + assertThat(delete.binds().get("operate_time")).isInstanceOf(Instant.class); + } +} diff --git a/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcSessionStoreTest.java b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcSessionStoreTest.java new file mode 100644 index 000000000..254b111be --- /dev/null +++ b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcSessionStoreTest.java @@ -0,0 +1,270 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package io.github.pnoker.db.agentic; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.github.pnoker.common.agentic.entity.bo.SessionBO; +import io.github.pnoker.common.agentic.entity.model.SessionExt; +import io.github.pnoker.common.entity.common.RequestHeader; +import io.github.pnoker.db.core.dialect.StandardR2dbcDialect; +import io.github.pnoker.db.core.transaction.PageTransaction; +import java.time.Instant; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import reactor.test.StepVerifier; +import tools.jackson.databind.ObjectMapper; + +/** + * Regression coverage for the session store concurrency contract: soft-deleted conversation ids + * revive instead of failing on the natural-key constraint, concurrent updates merge under the + * row lock instead of losing writes, the delete locks the session row before touching children, + * and every timestamp bind is an {@link Instant} so TIMESTAMPTZ encoding cannot drift with the + * database session timezone. + */ +class R2dbcSessionStoreTest { + + private static final long TENANT_ID = 7L; + private static final long USER_ID = 42L; + private static final String CONVERSATION_ID = "conv-1"; + + private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase(); + private final ObjectMapper objectMapper = new ObjectMapper(); + private final R2dbcSessionStore store = new R2dbcSessionStore( + database, + ScriptedAgenticDatabase.passThroughOperator(), + Mockito.mock(PageTransaction.class), + objectMapper, + new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint")); + + private RequestHeader.PrincipalHeader header; + + @BeforeEach + void setUp() { + header = new RequestHeader.PrincipalHeader(); + header.setTenantId(TENANT_ID); + header.setPrincipalId(USER_ID); + header.setPrincipalName("operator"); + } + + @Test + void touchRevivesASoftDeletedConversationAsAFreshSession() { + database.seedSession( + TENANT_ID, USER_ID, CONVERSATION_ID, "Old Title", "{\"model\":\"dead\",\"temperature\":0.1}", true); + SessionExt incoming = new SessionExt(); + incoming.setModel("fresh-model"); + + StepVerifier.create(store.touch(CONVERSATION_ID, incoming, header)) + .assertNext(session -> { + assertThat(session.getTitle()).isEqualTo("New Conversation"); + assertThat(session.getSessionExt().getModel()).isEqualTo("fresh-model"); + assertThat(session.getSessionExt().getTemperature()) + .as("the tombstone's dead preferences must not leak into the revived session") + .isNull(); + }) + .verifyComplete(); + + ScriptedAgenticDatabase.SessionRow row = database.session(TENANT_ID, USER_ID, CONVERSATION_ID); + assertThat(row.deleted).as("the tombstone row must be live again").isZero(); + assertThat(row.title).isEqualTo("New Conversation"); + assertThat((Instant) row.createTime) + .as("revive restarts the conversation lifecycle") + .isAfter(Instant.parse("2026-01-02T00:00:00Z")); + assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_session")) + .as("revive must not attempt an insert against the unique natural key") + .isZero(); + } + + @Test + void touchMergesIntoTheLiveSessionUnderLock() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Keep", "{\"temperature\":0.5}", false); + SessionExt incoming = new SessionExt(); + incoming.setModel("m1"); + + StepVerifier.create(store.touch(CONVERSATION_ID, incoming, header)) + .assertNext(session -> { + assertThat(session.getSessionExt().getModel()).isEqualTo("m1"); + assertThat(session.getSessionExt().getTemperature()).isEqualTo(0.5); + assertThat(session.getTitle()).isEqualTo("Keep"); + }) + .verifyComplete(); + + assertThat(database.firstMatching("LIMIT 1 FOR UPDATE").sql()) + .as("the decision must read the row under FOR UPDATE without a deleted filter") + .doesNotContain("deleted = 0"); + } + + @Test + void touchRecoversWhenAConcurrentInsertWinsTheConversation() { + database.scriptNextSessionInsertViolation(() -> + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Winner", "{\"temperature\":0.7}", false)); + SessionExt incoming = new SessionExt(); + incoming.setModel("mine"); + + StepVerifier.create(store.touch(CONVERSATION_ID, incoming, header)) + .assertNext(session -> { + assertThat(session.getSessionExt().getModel()).isEqualTo("mine"); + assertThat(session.getSessionExt().getTemperature()) + .as("the retry merges onto the concurrent winner's state") + .isEqualTo(0.7); + }) + .verifyComplete(); + + assertThat(database.session(TENANT_ID, USER_ID, CONVERSATION_ID).deleted) + .isZero(); + } + + @Test + void concurrentUpdatesMergeInsteadOfLosingTheFirstWrite() throws Exception { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Base", "{\"temperature\":0.5}", false); + // Park the first updater's session UPDATE mid-flight so the second updater subscribes + // while the first is still inside its locked read-merge-write window. + ScriptedAgenticDatabase.Hold firstUpdateHold = database.holdNext("UPDATE dc3_agentic.dc3_session"); + + SessionExt temperatureChange = new SessionExt(); + temperatureChange.setTemperature(0.9); + CompletableFuture first = store.update(CONVERSATION_ID, temperatureChange, "First Title", header) + .toFuture(); + firstUpdateHold.awaitRouted(); + + SessionExt modelChange = new SessionExt(); + modelChange.setModel("m2"); + CompletableFuture second = + store.update(CONVERSATION_ID, modelChange, null, header).toFuture(); + assertThat(second.isDone()) + .as("the second updater must wait on the row lock held by the first") + .isFalse(); + + firstUpdateHold.release(); + SessionBO firstResult = first.get(5, TimeUnit.SECONDS); + SessionBO secondResult = second.get(5, TimeUnit.SECONDS); + + assertThat(firstResult.getSessionExt().getTemperature()).isEqualTo(0.9); + assertThat(secondResult.getSessionExt().getModel()).isEqualTo("m2"); + assertThat(secondResult.getSessionExt().getTemperature()) + .as("the second update must merge over the first update's committed state, not lose it") + .isEqualTo(0.9); + assertThat(secondResult.getTitle()) + .as("a blank title keeps the first updater's rename") + .isEqualTo("First Title"); + } + + @Test + void updateRunsTheLockMergeWriteSequenceInsideOneTransaction() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Base", "{}", false); + StepVerifier.create(store.update(CONVERSATION_ID, null, "Renamed", header)) + .assertNext(session -> assertThat(session.getTitle()).isEqualTo("Renamed")) + .verifyComplete(); + + assertThat(database.indexOfFirst("deleted = 0 LIMIT 1 FOR UPDATE")) + .as("the update path must open with a locking read of the live row") + .isZero(); + assertThat(database.firstMatching("UPDATE dc3_agentic.dc3_session").binds()) + .containsEntry("title", "Renamed"); + } + + @Test + void updateCompletesEmptyWhenTheSessionDoesNotExist() { + StepVerifier.create(store.update("missing", null, "Renamed", header)).verifyComplete(); + assertThat(database.countMatching("UPDATE dc3_agentic.dc3_session")).isZero(); + } + + @Test + void deleteLocksTheLiveSessionBeforeSoftDeletingChildren() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Doomed", "{}", false); + database.seedAttachment(TENANT_ID, USER_ID, CONVERSATION_ID, "/store/a.txt", false); + database.seedAttachment(TENANT_ID, USER_ID, CONVERSATION_ID, "/store/b.txt", false); + + StepVerifier.create(store.delete(CONVERSATION_ID, header)) + .expectNext(1L) + .verifyComplete(); + + assertThat(database.indexOfFirst("SELECT id FROM dc3_agentic.dc3_session")) + .as("the delete must lock the session row first so uploads serialize against it") + .isZero(); + assertThat(database.session(TENANT_ID, USER_ID, CONVERSATION_ID).deleted) + .isOne(); + assertThat(database.attachments()) + .as("the delete must soft-delete the conversation's attachments") + .allSatisfy(row -> assertThat(row.deleted).isOne()); + List statements = database.executions(); + assertThat(statements.get(statements.size() - 1).sql()) + .as("the session row must be soft-deleted after the children") + .contains("UPDATE dc3_agentic.dc3_session SET deleted = 1"); + } + + @Test + void deleteIsANoOpWhenTheSessionIsAlreadyDeleted() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Gone", "{}", true); + + StepVerifier.create(store.delete(CONVERSATION_ID, header)) + .expectNext(0L) + .verifyComplete(); + + assertThat(database.countMatching("UPDATE dc3_agentic.dc3_session")).isZero(); + assertThat(database.countMatching("UPDATE dc3_agentic.dc3_message")).isZero(); + } + + @Test + void touchBindsTimestampsAsInstantsSoEncodingIsTimezoneIndependent() { + StepVerifier.create(store.touch(CONVERSATION_ID, null, header)) + .expectNextCount(1) + .verifyComplete(); + + ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_session"); + assertThat(insert.binds().get("create_time")) + .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock") + .isInstanceOf(Instant.class); + assertThat(insert.binds().get("operate_time")).isInstanceOf(Instant.class); + } + + @Test + void reviveAndDeleteBindsTimestampsAsInstants() { + database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Old", "{}", true); + StepVerifier.create(store.touch(CONVERSATION_ID, null, header)) + .expectNextCount(1) + .verifyComplete(); + assertThat(database.firstMatching("SET deleted = 0").binds().get("create_time")) + .isInstanceOf(Instant.class); + + StepVerifier.create(store.delete(CONVERSATION_ID, header)) + .expectNext(1L) + .verifyComplete(); + assertThat(database.firstMatching("UPDATE dc3_agentic.dc3_session SET deleted = 1") + .binds() + .get("operate_time")) + .isInstanceOf(Instant.class); + } + + @Test + void getNormalizesNonUtcDriverTimesToUtcWallClock() { + database.seedSessionWithOffsetTimes(TENANT_ID, USER_ID, CONVERSATION_ID); + + StepVerifier.create(store.get(CONVERSATION_ID, header)) + .assertNext(session -> { + assertThat(session.getCreateTime()) + .as("an offset driver decode must normalize to the UTC wall clock") + .isEqualTo(java.time.LocalDateTime.parse("2026-03-01T04:00")); + assertThat(session.getOperateTime()).isEqualTo(java.time.LocalDateTime.parse("2026-03-01T04:00")); + }) + .verifyComplete(); + } +} diff --git a/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/ScriptedAgenticDatabase.java b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/ScriptedAgenticDatabase.java new file mode 100644 index 000000000..fea43c04e --- /dev/null +++ b/dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/ScriptedAgenticDatabase.java @@ -0,0 +1,975 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package io.github.pnoker.db.agentic; + +import io.r2dbc.spi.ConnectionFactory; +import io.r2dbc.spi.Row; +import io.r2dbc.spi.RowMetadata; +import java.time.Instant; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Supplier; +import org.mockito.ArgumentMatchers; +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.r2dbc.core.DatabaseClient; +import org.springframework.r2dbc.core.FetchSpec; +import org.springframework.r2dbc.core.RowsFetchSpec; +import org.springframework.transaction.reactive.TransactionalOperator; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * In-memory stand-in for the agentic PostgreSQL schema, purpose-built for the store unit tests. + * + *

Routes the stores' explicit SQL by statement shape, keeps mutable session, attachment, and + * message rows, records every executed statement with its bind values, and models READ COMMITTED + * row locking: {@code SELECT ... FOR UPDATE} serializes per conversation (FIFO waiters that + * re-read row state once granted), a session-row mutation releases the lock, and a natural-key + * insert colliding with an existing row raises {@link DataIntegrityViolationException}. Tests can + * park one statement mid-flight with {@link #holdNext(String)} to force real interleavings. + */ +final class ScriptedAgenticDatabase implements DatabaseClient { + + private final Map sessions = new LinkedHashMap<>(); + private final List attachments = new ArrayList<>(); + private final List messages = new ArrayList<>(); + private final List actions = new ArrayList<>(); + private final Map locks = new LinkedHashMap<>(); + private final List executions = new ArrayList<>(); + private final ArrayDeque holds = new ArrayDeque<>(); + private final AtomicLong sessionIds = new AtomicLong(100); + private final AtomicLong attachmentIds = new AtomicLong(200); + private final AtomicLong messageIds = new AtomicLong(300); + private final AtomicLong actionIds = new AtomicLong(400); + private Runnable sessionInsertViolationScript; + + /** One recorded statement with its bind values and typed null binds. */ + record Execution(String sql, Map binds, Map> nulls) { + + boolean contains(String marker) { + return sql.contains(marker); + } + } + + /** A mutable soft-deletable {@code dc3_session} row. */ + static final class SessionRow { + Long id; + String conversationId; + String title; + String sessionExtJson; + Long tenantId; + Long userId; + Long creatorId; + String creatorName; + Object createTime; + Long operatorId; + String operatorName; + Object operateTime; + int deleted; + } + + /** A mutable soft-deletable {@code dc3_attachment} row. */ + static final class AttachmentRow { + Long id; + String conversationId; + String fileName; + String contentType; + Long size; + String filePath; + Long tenantId; + Long userId; + Long creatorId; + String creatorName; + Instant createTime; + Long operatorId; + String operatorName; + Instant operateTime; + int deleted; + } + + /** A mutable soft-deletable {@code dc3_message} row. */ + static final class MessageRow { + Long id; + String conversationId; + String role; + String content; + String model; + Long messageIndex; + Byte status; + Long tenantId; + Long userId; + Instant createTime; + int deleted; + } + + /** A mutable soft-deletable {@code dc3_action} row. */ + static final class ActionRow { + Long id; + String actionId; + String conversationId; + String actionType; + String title; + String description; + String payload; + Byte status; + Instant expireTime; + Long tenantId; + Long userId; + Instant createTime; + int deleted; + } + + /** Park the next statement whose SQL contains the marker until {@link Hold#release()}. */ + Hold holdNext(String marker) { + Hold hold = new Hold(marker); + synchronized (holds) { + holds.add(hold); + } + return hold; + } + + /** Fail the next session insert with a unique-key violation, running the script first. */ + void scriptNextSessionInsertViolation(Runnable script) { + sessionInsertViolationScript = script; + } + + List executions() { + synchronized (executions) { + return List.copyOf(executions); + } + } + + Execution firstMatching(String marker) { + return executions().stream() + .filter(execution -> execution.contains(marker)) + .findFirst() + .orElseThrow(() -> new AssertionError("no statement containing: " + marker)); + } + + int countMatching(String marker) { + return (int) executions().stream() + .filter(execution -> execution.contains(marker)) + .count(); + } + + int indexOfFirst(String marker) { + List all = executions(); + for (int index = 0; index < all.size(); index++) { + if (all.get(index).contains(marker)) return index; + } + return -1; + } + + void seedSession( + long tenantId, long userId, String conversationId, String title, String sessionExtJson, boolean deleted) { + SessionRow row = new SessionRow(); + row.id = sessionIds.incrementAndGet(); + row.tenantId = tenantId; + row.userId = userId; + row.conversationId = conversationId; + row.title = title; + row.sessionExtJson = sessionExtJson; + row.creatorId = userId; + row.creatorName = "seeder"; + row.operatorId = userId; + row.operatorName = "seeder"; + row.createTime = Instant.parse("2026-01-01T00:00:00Z"); + row.operateTime = row.createTime; + row.deleted = deleted ? 1 : 0; + sessions.put(key(tenantId, userId, conversationId), row); + } + + /** Seed a session whose decoded times carry a non-UTC offset, like a driver in a non-UTC session timezone. */ + void seedSessionWithOffsetTimes(long tenantId, long userId, String conversationId) { + seedSession(tenantId, userId, conversationId, "Offset", "{}", false); + SessionRow row = sessions.get(key(tenantId, userId, conversationId)); + row.createTime = java.time.OffsetDateTime.parse("2026-03-01T12:00:00+08:00"); + row.operateTime = java.time.OffsetDateTime.parse("2026-03-01T12:00:00+08:00"); + } + + void seedAttachment(long tenantId, long userId, String conversationId, String filePath, boolean deleted) { + AttachmentRow row = new AttachmentRow(); + row.id = attachmentIds.incrementAndGet(); + row.tenantId = tenantId; + row.userId = userId; + row.conversationId = conversationId; + row.fileName = "seed.txt"; + row.contentType = "text/plain"; + row.size = 1L; + row.filePath = filePath; + row.creatorId = userId; + row.createTime = Instant.parse("2026-01-01T00:00:00Z"); + row.deleted = deleted ? 1 : 0; + attachments.add(row); + } + + SessionRow session(long tenantId, long userId, String conversationId) { + return sessions.get(key(tenantId, userId, conversationId)); + } + + List attachments() { + return attachments; + } + + List messages() { + return messages; + } + + /** A transactional operator mock that passes publishers through unchanged. */ + static TransactionalOperator passThroughOperator() { + TransactionalOperator operator = org.mockito.Mockito.mock(TransactionalOperator.class); + org.mockito.Mockito.when(operator.transactional(ArgumentMatchers.>any())) + .thenAnswer(invocation -> invocation.getArgument(0)); + org.mockito.Mockito.when(operator.transactional(ArgumentMatchers.>any())) + .thenAnswer(invocation -> invocation.getArgument(0)); + return operator; + } + + /** A page transaction mock that passes publishers through unchanged. */ + static io.github.pnoker.db.core.transaction.PageTransaction passThroughPageTransaction() { + io.github.pnoker.db.core.transaction.PageTransaction transaction = + org.mockito.Mockito.mock(io.github.pnoker.db.core.transaction.PageTransaction.class); + org.mockito.Mockito.when(transaction.transactional(ArgumentMatchers.>any())) + .thenAnswer(invocation -> invocation.getArgument(0)); + return transaction; + } + + @Override + public GenericExecuteSpec sql(String sql) { + return new ScriptedSpec(sql); + } + + @Override + public GenericExecuteSpec sql(Supplier sqlSupplier) { + return new ScriptedSpec(sqlSupplier.get()); + } + + @Override + public ConnectionFactory getConnectionFactory() { + throw new UnsupportedOperationException("not used by the store unit tests"); + } + + @Override + public Mono inConnection(java.util.function.Function> connectionFunction) { + throw new UnsupportedOperationException("not used by the store unit tests"); + } + + @Override + public Flux inConnectionMany( + java.util.function.Function> connectionFunction) { + throw new UnsupportedOperationException("not used by the store unit tests"); + } + + /** Parked-statement handle; the routed latch proves the statement reached the database. */ + static final class Hold { + private final String marker; + private final CompletableFuture gate = new CompletableFuture<>(); + private final CountDownLatch routed = new CountDownLatch(1); + + private Hold(String marker) { + this.marker = marker; + } + + void awaitRouted() throws InterruptedException { + if (!routed.await(5, java.util.concurrent.TimeUnit.SECONDS)) { + throw new AssertionError("held statement never routed: " + marker); + } + } + + void release() { + gate.complete(null); + } + } + + private static String key(Long tenantId, Long userId, String conversationId) { + return tenantId + "/" + userId + "/" + conversationId; + } + + private static String key(Map binds) { + return key((Long) binds.get("tenant_id"), (Long) binds.get("user_id"), (String) binds.get("conversation_id")); + } + + private static final class LockBox { + private boolean held; + private final ArrayDeque> waiters = new ArrayDeque<>(); + + synchronized boolean tryAcquire() { + if (held) return false; + held = true; + return true; + } + + synchronized CompletableFuture enqueue() { + CompletableFuture granted = new CompletableFuture<>(); + waiters.add(granted); + return granted; + } + + synchronized void release() { + CompletableFuture next = waiters.poll(); + if (next == null) { + held = false; + } else { + next.complete(null); + } + } + } + + private Mono acquire(String lockKey) { + return Mono.defer(() -> { + LockBox box = locks.computeIfAbsent(lockKey, ignored -> new LockBox()); + if (box.tryAcquire()) return Mono.empty(); + return Mono.fromFuture(box.enqueue()).then(); + }); + } + + private void release(String lockKey) { + LockBox box = locks.get(lockKey); + if (box != null) box.release(); + } + + private record Outcome(List> rows, long rowsUpdated) { + + static final Outcome EMPTY = new Outcome(List.of(), 0); + } + + private final class ScriptedSpec implements GenericExecuteSpec { + + private final String sql; + private final Map binds = new LinkedHashMap<>(); + private final Map> nulls = new LinkedHashMap<>(); + + private ScriptedSpec(String sql) { + this.sql = sql; + } + + @Override + public GenericExecuteSpec bind(int index, Object value) { + throw new UnsupportedOperationException("positional binds are not used"); + } + + @Override + public GenericExecuteSpec bindNull(int index, Class type) { + throw new UnsupportedOperationException("positional binds are not used"); + } + + @Override + public GenericExecuteSpec bind(String name, Object value) { + binds.put(name, value); + return this; + } + + @Override + public GenericExecuteSpec bindNull(String name, Class type) { + nulls.put(name, type); + return this; + } + + @Override + public GenericExecuteSpec bindValues(List values) { + throw new UnsupportedOperationException("not used"); + } + + @Override + public GenericExecuteSpec bindValues(Map values) { + binds.putAll(values); + return this; + } + + @Override + public GenericExecuteSpec bindProperties(Object source) { + throw new UnsupportedOperationException("not used"); + } + + @Override + public GenericExecuteSpec filter(org.springframework.r2dbc.core.StatementFilterFunction filter) { + throw new UnsupportedOperationException("not used"); + } + + @Override + public GenericExecuteSpec filter( + java.util.function.Function filter) { + throw new UnsupportedOperationException("not used"); + } + + @Override + public RowsFetchSpec map(java.util.function.Function mapping) { + return mapped((row, ignored) -> mapping.apply(row)); + } + + @Override + public RowsFetchSpec map(java.util.function.BiFunction mapping) { + return mapped(mapping); + } + + @Override + public RowsFetchSpec mapValue(Class type) { + throw new UnsupportedOperationException("not used"); + } + + @Override + public RowsFetchSpec mapProperties(Class type) { + throw new UnsupportedOperationException("not used"); + } + + @Override + public Flux flatMap( + java.util.function.Function> mapping) { + throw new UnsupportedOperationException("not used"); + } + + @Override + public FetchSpec> fetch() { + return new ScriptedFetchSpec(this); + } + + @Override + public Mono then() { + return resolveOutcome().then(); + } + + /** Resolve this statement: record it, honor any hold, then route and apply effects. */ + Mono resolveOutcome() { + return Mono.defer(() -> { + Hold hold = takeHold(sql); + Mono body = Mono.defer(() -> route(sql, binds)); + if (hold == null) { + return body; + } + return Mono.fromFuture(hold.gate).then(body); + }) + .doOnSubscribe(subscription -> record(sql, binds, nulls)); + } + + private RowsFetchSpec mapped(java.util.function.BiFunction mapping) { + return new RowsFetchSpec<>() { + @Override + public Mono one() { + return resolveOutcome() + .flatMap(outcome -> outcome.rows().isEmpty() + ? Mono.empty() + : Mono.just(mapping.apply( + new FakeRow(outcome.rows().get(0)), (RowMetadata) null))); + } + + @Override + public Mono first() { + return one(); + } + + @Override + public Flux all() { + return resolveOutcome() + .flatMapMany(outcome -> Flux.fromIterable(outcome.rows().stream() + .map(row -> mapping.apply(new FakeRow(row), null)) + .toList())); + } + }; + } + + private Hold takeHold(String statement) { + synchronized (holds) { + for (Hold hold : holds) { + if (statement.contains(hold.marker)) { + holds.remove(hold); + hold.routed.countDown(); + return hold; + } + } + } + return null; + } + + private void record(String statement, Map values, Map> typedNulls) { + synchronized (executions) { + executions.add(new Execution(statement, Map.copyOf(values), Map.copyOf(typedNulls))); + } + } + } + + private final class ScriptedFetchSpec implements FetchSpec> { + + private final ScriptedSpec spec; + + private ScriptedFetchSpec(ScriptedSpec spec) { + this.spec = spec; + } + + @Override + public Mono> one() { + return resolveRows().flatMap(rows -> rows.isEmpty() ? Mono.empty() : Mono.just(rows.get(0))); + } + + @Override + public Mono> first() { + return one(); + } + + @Override + public Flux> all() { + return resolveRows().flatMapMany(Flux::fromIterable); + } + + @Override + public Mono rowsUpdated() { + return spec.resolveOutcome().map(Outcome::rowsUpdated); + } + + private Mono>> resolveRows() { + return spec.resolveOutcome().map(Outcome::rows); + } + } + + private Mono route(String sql, Map binds) { + if (sql.startsWith("SELECT id FROM dc3_agentic.dc3_session")) { + return lockSessionId(binds); + } + if (sql.startsWith("INSERT INTO dc3_agentic.dc3_session")) { + return insertSession(binds); + } + if (sql.startsWith("UPDATE dc3_agentic.dc3_session")) { + return updateSession(sql, binds); + } + if (sql.startsWith("SELECT id, conversation_id, title, session_ext")) { + return selectSession(sql, binds); + } + if (sql.startsWith("INSERT INTO dc3_agentic.dc3_attachment")) { + return insertAttachment(binds); + } + if (sql.startsWith("SELECT file_path FROM dc3_agentic.dc3_attachment")) { + return listDeletedFilePaths(binds); + } + if (sql.startsWith("SELECT id, conversation_id, file_name")) { + return selectAttachment(binds); + } + if (sql.startsWith("UPDATE dc3_agentic.dc3_attachment SET deleted = 1")) { + return softDeleteAttachments(binds); + } + if (sql.startsWith("UPDATE dc3_agentic.dc3_message SET deleted = 1")) { + return softDeleteMessages(binds); + } + if (sql.startsWith("UPDATE dc3_agentic.dc3_action SET deleted = 1")) { + return Mono.just(new Outcome(List.of(), 0)); + } + if (sql.startsWith("INSERT INTO dc3_agentic.dc3_action")) { + return insertAction(binds); + } + if (sql.startsWith("UPDATE dc3_agentic.dc3_action")) { + return updateAction(binds); + } + if (sql.startsWith("INSERT INTO dc3_agentic.dc3_message")) { + return insertMessage(binds); + } + if (sql.startsWith("SELECT COALESCE(MAX(message_index)")) { + return nextMessageIndex(binds); + } + if (sql.startsWith("SELECT id, action_id, conversation_id")) { + return selectAction(binds); + } + if (sql.startsWith("SELECT COUNT(*) AS total")) { + return Mono.just(new Outcome(List.of(Map.of("total", 0L)), 0)); + } + if (sql.startsWith("SELECT id, conversation_id, role, content")) { + return selectMessage(binds); + } + // Statements outside the tested protocol (model config updates) still execute and record + // their binds; they simply report one affected row and no data. + return Mono.just(new Outcome(List.of(), 1)); + } + + private Mono lockSessionId(Map binds) { + // FOR UPDATE on no matching row locks nothing, so an absent session must not queue. + if (liveSession(binds) == null) return Mono.just(Outcome.EMPTY); + String lockKey = key(binds); + return acquire(lockKey).then(Mono.fromCallable(() -> { + SessionRow row = liveSession(binds); + return row == null ? Outcome.EMPTY : new Outcome(List.of(Map.of("id", row.id)), 0); + })); + } + + private Mono insertSession(Map binds) { + Runnable script = sessionInsertViolationScript; + if (script != null) { + sessionInsertViolationScript = null; + script.run(); + if (sessions.containsKey(key(binds))) { + return Mono.error(new DataIntegrityViolationException("uq_session_tenant_user_conversation")); + } + } + if (sessions.containsKey(key(binds))) { + return Mono.error(new DataIntegrityViolationException("uq_session_tenant_user_conversation")); + } + SessionRow row = new SessionRow(); + row.id = sessionIds.incrementAndGet(); + row.tenantId = (Long) binds.get("tenant_id"); + row.userId = (Long) binds.get("user_id"); + row.conversationId = (String) binds.get("conversation_id"); + row.title = (String) binds.get("title"); + row.sessionExtJson = (String) binds.get("session_ext"); + row.creatorId = (Long) binds.get("creator_id"); + row.operatorId = (Long) binds.get("operator_id"); + row.createTime = (Instant) binds.get("create_time"); + row.operateTime = (Instant) binds.get("operate_time"); + row.deleted = 0; + sessions.put(key(row.tenantId, row.userId, row.conversationId), row); + release(key(binds)); + return Mono.just(new Outcome(List.of(), 1)); + } + + private Mono updateSession(String sql, Map binds) { + if (sql.contains("SET deleted = 1")) { + SessionRow row = liveSession(binds); + if (row == null) return Mono.just(Outcome.EMPTY); + row.deleted = 1; + row.operatorId = (Long) binds.get("operator_id"); + row.operateTime = (Instant) binds.get("operate_time"); + release(key(row.tenantId, row.userId, row.conversationId)); + return Mono.just(new Outcome(List.of(), 1)); + } + SessionRow row = sessionById((Long) binds.get("id")); + // The plain update's WHERE clause guards on deleted = 0; the revive's SET clause clears + // the tombstone flag, so distinguish by the statement tail, not by a bare substring. + if (row == null || (sql.endsWith("deleted = 0") && row.deleted != 0)) { + return Mono.just(Outcome.EMPTY); + } + if (sql.contains("SET deleted = 0")) { + row.deleted = 0; + row.title = (String) binds.get("title"); + row.sessionExtJson = (String) binds.get("session_ext"); + row.creatorId = (Long) binds.get("creator_id"); + row.creatorName = (String) binds.get("creator_name"); + row.createTime = (Instant) binds.get("create_time"); + row.operatorId = (Long) binds.get("operator_id"); + row.operateTime = (Instant) binds.get("operate_time"); + } else { + row.title = (String) binds.get("title"); + row.sessionExtJson = (String) binds.get("session_ext"); + row.operatorId = (Long) binds.get("operator_id"); + row.operateTime = (Instant) binds.get("operate_time"); + } + release(key(row.tenantId, row.userId, row.conversationId)); + return Mono.just(new Outcome(List.of(), 1)); + } + + private Mono selectSession(String sql, Map binds) { + boolean liveOnly = sql.contains("AND deleted = 0"); + if (sql.contains("FOR UPDATE")) { + // FOR UPDATE on no matching row locks nothing, so an absent row must not queue. + if ((liveOnly ? liveSession(binds) : anySession(binds)) == null) { + return Mono.just(Outcome.EMPTY); + } + String lockKey = key(binds); + return acquire(lockKey).then(Mono.fromCallable(() -> { + SessionRow row = liveOnly ? liveSession(binds) : anySession(binds); + return row == null ? Outcome.EMPTY : new Outcome(List.of(sessionMap(row)), 0); + })); + } + SessionRow row = liveSession(binds); + return Mono.just(row == null ? Outcome.EMPTY : new Outcome(List.of(sessionMap(row)), 0)); + } + + private Mono insertAttachment(Map binds) { + AttachmentRow row = new AttachmentRow(); + row.id = attachmentIds.incrementAndGet(); + row.tenantId = (Long) binds.get("tenant_id"); + row.userId = (Long) binds.get("user_id"); + row.conversationId = (String) binds.get("conversation_id"); + row.fileName = (String) binds.get("file_name"); + row.contentType = (String) binds.get("content_type"); + row.size = (Long) binds.get("size"); + row.filePath = (String) binds.get("file_path"); + row.creatorId = (Long) binds.get("creator_id"); + row.creatorName = (String) binds.get("creator_name"); + row.createTime = (Instant) binds.get("create_time"); + row.operatorId = (Long) binds.get("operator_id"); + row.operateTime = (Instant) binds.get("operate_time"); + row.deleted = 0; + attachments.add(row); + release(key(binds)); + return Mono.just(new Outcome(List.of(), 1)); + } + + private Mono listDeletedFilePaths(Map binds) { + List> rows = attachments.stream() + .filter(row -> matchesConversation(row, binds) && row.deleted == 1) + .map(row -> Map.of("file_path", row.filePath)) + .toList(); + return Mono.just(new Outcome(rows, 0)); + } + + private Mono selectAttachment(Map binds) { + if (binds.containsKey("file_path")) { + AttachmentRow found = attachments.stream() + .filter(row -> row.filePath.equals(binds.get("file_path")) + && row.tenantId.equals(binds.get("tenant_id")) + && row.userId.equals(binds.get("user_id")) + && row.deleted == 0) + .findFirst() + .orElse(null); + return Mono.just(found == null ? Outcome.EMPTY : new Outcome(List.of(attachmentMap(found)), 0)); + } + List> rows = attachments.stream() + .filter(row -> matchesConversation(row, binds) && row.deleted == 0) + .map(this::attachmentMap) + .toList(); + return Mono.just(new Outcome(rows, 0)); + } + + private Mono softDeleteAttachments(Map binds) { + long affected = attachments.stream() + .filter(row -> matchesConversation(row, binds) && row.deleted == 0) + .peek(row -> row.deleted = 1) + .count(); + return Mono.just(new Outcome(List.of(), affected)); + } + + private Mono softDeleteMessages(Map binds) { + long affected = messages.stream() + .filter(row -> matchesConversation(row, binds) && row.deleted == 0) + .peek(row -> row.deleted = 1) + .count(); + return Mono.just(new Outcome(List.of(), affected)); + } + + private Mono insertMessage(Map binds) { + MessageRow row = new MessageRow(); + row.id = messageIds.incrementAndGet(); + row.tenantId = (Long) binds.get("tenant_id"); + row.userId = (Long) binds.get("user_id"); + row.conversationId = (String) binds.get("conversation_id"); + row.role = (String) binds.get("role"); + row.content = (String) binds.get("content"); + row.model = (String) binds.get("model"); + row.messageIndex = (Long) binds.get("message_index"); + row.status = (Byte) binds.get("status"); + row.createTime = (Instant) binds.get("create_time"); + row.deleted = 0; + messages.add(row); + release(key(binds)); + return Mono.just(new Outcome(List.of(), 1)); + } + + private Mono nextMessageIndex(Map binds) { + long next = messages.stream() + .filter(row -> matchesConversation(row, binds) && row.deleted == 0) + .mapToLong(row -> row.messageIndex) + .max() + .orElse(-1L) + + 1; + return Mono.just(new Outcome(List.of(Map.of("next_index", next)), 0)); + } + + private Mono selectMessage(Map binds) { + MessageRow found = messages.stream() + .filter(row -> matchesConversation(row, binds) + && row.deleted == 0 + && row.messageIndex.equals(binds.get("message_index"))) + .findFirst() + .orElse(null); + if (found == null) return Mono.just(Outcome.EMPTY); + Map row = new LinkedHashMap<>(); + row.put("id", found.id); + row.put("conversation_id", found.conversationId); + row.put("role", found.role); + row.put("content", found.content); + row.put("model", found.model); + row.put("message_index", found.messageIndex); + row.put("status", found.status); + row.put("tenant_id", found.tenantId); + row.put("user_id", found.userId); + row.put("remark", ""); + row.put("creator_id", found.tenantId); + row.put("creator_name", ""); + row.put("create_time", found.createTime); + row.put("operator_id", found.tenantId); + row.put("operator_name", ""); + row.put("operate_time", found.createTime); + return Mono.just(new Outcome(List.of(row), 0)); + } + + private SessionRow liveSession(Map binds) { + SessionRow row = anySession(binds); + return row != null && row.deleted == 0 ? row : null; + } + + private Mono insertAction(Map binds) { + ActionRow row = new ActionRow(); + row.id = actionIds.incrementAndGet(); + row.actionId = (String) binds.get("action_id"); + row.conversationId = (String) binds.get("conversation_id"); + row.actionType = (String) binds.get("action_type"); + row.title = (String) binds.get("title"); + row.description = (String) binds.get("description"); + row.payload = (String) binds.get("payload"); + row.status = (Byte) binds.get("status"); + row.tenantId = (Long) binds.get("tenant_id"); + row.userId = (Long) binds.get("user_id"); + row.createTime = (Instant) binds.get("create_time"); + row.expireTime = (Instant) binds.get("expire_time"); + row.deleted = 0; + actions.add(row); + return Mono.just(new Outcome(List.of(), 1)); + } + + private Mono updateAction(Map binds) { + ActionRow row = actions.stream() + .filter(candidate -> candidate.actionId.equals(binds.get("action_id")) + && candidate.tenantId.equals(binds.get("tenant_id")) + && candidate.userId.equals(binds.get("user_id")) + && candidate.deleted == 0) + .findFirst() + .orElse(null); + if (row == null) return Mono.just(Outcome.EMPTY); + if (binds.containsKey("status")) row.status = (Byte) binds.get("status"); + return Mono.just(new Outcome(List.of(), 1)); + } + + private Mono selectAction(Map binds) { + ActionRow found = actions.stream() + .filter(candidate -> candidate.actionId.equals(binds.get("action_id")) + && candidate.tenantId.equals(binds.get("tenant_id")) + && candidate.userId.equals(binds.get("user_id")) + && candidate.deleted == 0) + .findFirst() + .orElse(null); + if (found == null) return Mono.just(Outcome.EMPTY); + Map row = new LinkedHashMap<>(); + row.put("id", found.id); + row.put("action_id", found.actionId); + row.put("conversation_id", found.conversationId); + row.put("action_type", found.actionType); + row.put("title", found.title); + row.put("description", found.description); + row.put("payload", found.payload); + row.put("status", found.status); + row.put("expire_time", found.expireTime); + row.put("tenant_id", found.tenantId); + row.put("user_id", found.userId); + row.put("remark", ""); + row.put("creator_id", found.tenantId); + row.put("creator_name", ""); + row.put("create_time", found.createTime); + row.put("operator_id", found.tenantId); + row.put("operator_name", ""); + row.put("operate_time", found.createTime); + return Mono.just(new Outcome(List.of(row), 0)); + } + + private SessionRow anySession(Map binds) { + return sessions.get(key(binds)); + } + + private SessionRow sessionById(Long id) { + return sessions.values().stream() + .filter(row -> row.id.equals(id)) + .findFirst() + .orElse(null); + } + + private boolean matchesConversation(AttachmentRow row, Map binds) { + return row.conversationId.equals(binds.get("conversation_id")) + && row.tenantId.equals(binds.get("tenant_id")) + && row.userId.equals(binds.get("user_id")); + } + + private boolean matchesConversation(MessageRow row, Map binds) { + return row.conversationId.equals(binds.get("conversation_id")) + && row.tenantId.equals(binds.get("tenant_id")) + && row.userId.equals(binds.get("user_id")); + } + + private Map sessionMap(SessionRow row) { + Map values = new LinkedHashMap<>(); + values.put("id", row.id); + values.put("conversation_id", row.conversationId); + values.put("title", row.title); + values.put("session_ext", row.sessionExtJson); + values.put("tenant_id", row.tenantId); + values.put("user_id", row.userId); + values.put("remark", ""); + values.put("creator_id", row.creatorId); + values.put("creator_name", row.creatorName); + values.put("create_time", row.createTime); + values.put("operator_id", row.operatorId); + values.put("operator_name", row.operatorName); + values.put("operate_time", row.operateTime); + values.put("deleted", row.deleted); + return values; + } + + private Map attachmentMap(AttachmentRow row) { + Map values = new LinkedHashMap<>(); + values.put("id", row.id); + values.put("conversation_id", row.conversationId); + values.put("file_name", row.fileName); + values.put("content_type", row.contentType); + values.put("size", row.size); + values.put("file_path", row.filePath); + values.put("tenant_id", row.tenantId); + values.put("user_id", row.userId); + values.put("remark", ""); + values.put("creator_id", row.creatorId); + values.put("creator_name", row.creatorName); + values.put("create_time", row.createTime); + values.put("operator_id", row.operatorId); + values.put("operator_name", row.operatorName); + values.put("operate_time", row.operateTime); + values.put("deleted", row.deleted); + return values; + } + + /** Minimal map-backed {@link Row} with the numeric coercions the store mappers rely on. */ + private static final class FakeRow implements Row { + + private final Map values; + + private FakeRow(Map values) { + this.values = values; + } + + @Override + public T get(String name, Class type) { + Object value = values.get(name); + if (value == null) return null; + if (type.isInstance(value)) return type.cast(value); + if (value instanceof Number number) { + if (type == Long.class) return type.cast(number.longValue()); + if (type == Integer.class) return type.cast(number.intValue()); + if (type == Double.class) return type.cast(number.doubleValue()); + if (type == Number.class) return type.cast(number); + } + if (type == String.class) return type.cast(value.toString()); + throw new IllegalStateException("unsupported column type: " + name); + } + + @Override + public T get(int index, Class type) { + throw new UnsupportedOperationException("not used"); + } + + @Override + public RowMetadata getMetadata() { + throw new UnsupportedOperationException("not used"); + } + } +} diff --git a/dc3/bin/check_r2dbc_migration.py b/dc3/bin/check_r2dbc_migration.py index 9ccdc45eb..d49cefb0d 100755 --- a/dc3/bin/check_r2dbc_migration.py +++ b/dc3/bin/check_r2dbc_migration.py @@ -115,6 +115,11 @@ def is_expected_startup_block(path: Path, category: str, match: re.Match[str]) - # bounded elastic scheduler so neither bridge ever runs on the event loop. if path.name == "SystemTool.java" and match.group(0).startswith(".block"): return True + # Attachment file cleanup deletes files with blocking NIO; the cleaner owns the + # bounded-elastic offload for every unlink so callers (session delete, upload + # rollback) can never run file deletion on an event loop. + if path.name == "AttachmentFileCleaner.java" and match.group(0).startswith("Schedulers.boundedElastic"): + return True return ( path.name == "OpenAiCompatibleAgenticRuntime.java" and match.group(0).startswith("Schedulers.boundedElastic")