From 9f1873735643c72f67eac5567fac7d49c8d2f20d Mon Sep 17 00:00:00 2001 From: pnoker Date: Tue, 20 Oct 2026 23:34:36 +0800 Subject: [PATCH 1/2] feat(agentic): add the hardening configuration surface Frozen property contract shared by the agentic audit remediation workstreams: agent loop termination policy, streaming backpressure budgets, provider key encryption master key, transport limiting timeouts/quotas, cache TTLs, connectivity cache bounds and attachment whitelist/quota settings. Co-Authored-By: Claude Code --- .../agentic/config/AgenticProperties.java | 134 ++++++++++++++++++ 1 file changed, 134 insertions(+) diff --git a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/config/AgenticProperties.java b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/config/AgenticProperties.java index 3ef97ab68..8dfd8cfd2 100644 --- a/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/config/AgenticProperties.java +++ b/dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/config/AgenticProperties.java @@ -138,4 +138,138 @@ public class AgenticProperties { */ @Min(value = 1, message = "Check max models must be greater than 0") private int checkMaxModels = 100; + + // ===== Agent loop policy ===== + + /** + * Maximum agent loop rounds (model call + tool executions) per chat turn. + * When the loop would exceed this bound the final round is replayed without + * tools so the model is forced to produce a textual answer instead of the + * turn failing hard, unless {@link #loopForceFinalRound} is disabled. + */ + @Min(value = 1, message = "Loop max rounds must be greater than 0") + private int loopMaxRounds = 8; + + /** + * Whether the last permitted loop round is replayed without tools to force a + * final textual answer. When false, exceeding {@link #loopMaxRounds} fails + * the turn (legacy behavior). + */ + private boolean loopForceFinalRound = true; + + // ===== Streaming backpressure policy ===== + + /** + * Frame budget the stream may buffer ahead of downstream demand before the + * provider side is considered stalled. A slow SSE consumer is tolerated up + * to this many buffered frames; the turn only fails once the buffer is full + * AND demand stays absent for {@link #streamIdleTimeoutSeconds}. + */ + @Min(value = 1, message = "Stream backpressure buffer must be greater than 0") + private int streamBackpressureBuffer = 256; + + /** + * Grace period in seconds during which a stream may sit with a full + * backpressure buffer and zero downstream demand before failing the turn. + * Shields long generations from transient consumer stalls (tab background, + * slow proxies) without unbounded memory growth. + */ + @Min(value = 1, message = "Stream idle timeout must be greater than 0 seconds") + private int streamIdleTimeoutSeconds = 60; + + /** + * Interval in seconds between SSE comment heartbeats emitted while a stream + * is open, keeping intermediaries from closing idle connections and making + * long tool rounds visible to consumers. Zero disables heartbeats. + */ + @Min(value = 0, message = "Stream heartbeat must not be negative") + private int streamHeartbeatSeconds = 15; + + // ===== Provider key encryption ===== + + /** + * Base64-encoded 256-bit master key used for AES-GCM envelope encryption of + * provider API keys at rest. When blank, a stable key is derived from the + * platform security key so local development keeps working; production must + * set an explicit key (environment variable DC3_AGENTIC_MASTER_KEY). + */ + private String securityMasterKey = ""; + + // ===== Transport hardening ===== + + /** + * Maximum seconds a chat request may wait for an in-flight slot before + * failing with an UNAVAILABLE problem instead of queueing forever. + */ + @Min(value = 1, message = "Transport acquire timeout must be greater than 0 seconds") + private int transportAcquireTimeoutSeconds = 30; + + /** + * Upper bound of concurrently in-flight model requests across ALL providers + * for paths that have no provider-specific semaphore (fallback model, + * uncached model configs). Protects the process when per-provider limits + * cannot apply. + */ + @Min(value = 1, message = "Transport global max concurrent requests must be greater than 0") + private int transportGlobalMaxConcurrentRequests = 16; + + /** + * Whether a tenant without a model configuration silently falls back to the + * platform-wide provider. Disabled by default: the fallback bypasses + * per-tenant accounting, so it must be an explicit deployment decision. + */ + private boolean transportFallbackEnabled = false; + + /** + * Per-tenant chat requests per second limit enforced at the chat entry + * point. Zero disables the limiter. + */ + @Min(value = 0, message = "Per-tenant QPS must not be negative") + private int transportPerTenantQps = 0; + + /** + * Time-to-live in seconds for the in-process provider/model/client caches. + * Bound stale entries after out-of-band provider edits in multi-replica + * deployments; eviction notices from the local node refresh immediately. + */ + @Min(value = 1, message = "Transport cache TTL must be greater than 0 seconds") + private int transportCacheTtlSeconds = 300; + + /** + * Upper bound of cached connectivity check results. Draft probes (no + * provider id) use a small fixed partition of this budget so they cannot + * grow the cache unboundedly. + */ + @Min(value = 1, message = "Check cache maximum size must be greater than 0") + private int checkCacheMaximumSize = 256; + + // ===== Attachment hardening ===== + + /** + * Comma-separated whitelist of attachment file extensions accepted for + * upload. Anything outside the list is rejected before any bytes are + * written. + */ + private String attachmentAllowedExtensions = "png,jpg,jpeg,gif,webp,csv,txt,json,md,pdf"; + + /** + * Maximum number of stored attachments per tenant. Uploads beyond the + * quota are rejected. + */ + @Min(value = 1, message = "Attachment tenant max count must be greater than 0") + private int attachmentTenantMaxCount = 2000; + + /** + * Maximum total stored attachment bytes per tenant. + */ + @Min(value = 1, message = "Attachment tenant max bytes must be greater than 0") + private long attachmentTenantMaxBytes = 1073741824L; + + /** + * Maximum length of an attachment file name accepted into prompts and + * storage metadata; longer names are truncated. Part of the untrusted + * content delimiting policy. + */ + @Min(value = 16, message = "Attachment name max length must be at least 16") + private int attachmentNameMaxLength = 128; } From 20cf63ef610ce41398b50b6ad2aa532ce73a32c3 Mon Sep 17 00:00:00 2001 From: pnoker Date: Wed, 21 Oct 2026 01:13:06 +0800 Subject: [PATCH 2/2] fix(agentic): make the store layer concurrent-safe and layered M21 (soft-delete tombstone poisons conversationId): touch() hit the full-table unique constraint uq_session_tenant_user_conversation when a soft-deleted row still held the natural key and surfaced the raw DataIntegrityViolationException. The constraint cannot become a partial unique index: dc3_message, dc3_attachment, and dc3_action reference it with foreign keys, and PostgreSQL FK targets must be a non-partial unique constraint/index. Product semantics are implemented in the write path instead: touch() now locks the row unfiltered (SELECT ... FOR UPDATE without a deleted filter), revives a tombstone as a brand-new session (deleted=0, title and session_ext reset, creator and create_time restarted), and retries the whole decision once when a concurrent insert wins the unique-key race. No DDL change, no fingerprint churn. M22 (session_ext lost update): update() read-merged-wrote without a lock or transaction, so concurrent renames overwrote each other. It now runs lock-merge-update inside one transaction, reusing touch()'s row lock, so concurrent preference changes merge. L51 (attachment save TOCTOU): R2dbcAttachmentStore.save inserted without checking the session, so a save racing a concurrent session delete survived as a live attachment row of a soft-deleted conversation. save now takes the message store's live-session guard (SELECT id FROM dc3_session ... deleted = 0 FOR UPDATE) in the same transaction, and the session delete locks the session row before soft-deleting children, so upload and delete serialize: the upload either lands before the delete or fails its guard. File cleanup is collected after commit from the new listDeletedFilePaths port query (soft-deleted rows only, so a revived conversation keeps its new files). M17 (layering inversion): R2dbcSessionStore injected the service-layer AttachmentFileCleaner and performed filesystem unlinking from the persistence adapter. The orchestration moved to SessionServiceImpl: the store is a pure SQL adapter again, the service reads the deleted attachment paths after the delete commits and unlinks best-effort. A layering guard test keeps service.impl imports out of the db module. M23 (blocking file IO on the event loop): AttachmentFileCleaner ran Files.deleteIfExists on the caller's thread, which is the Reactor Netty event loop after R2DBC completions. The cleaner now owns the offload: every unlink/unlinkAll subscribes on bounded elastic, with a documented exception in the r2dbc migration gate for the one boundedElastic reference. AttachmentServiceImpl's open/close paths are left for the security theater to align. M24 (timezone drift): the agentic stores bound UTC-naive LocalDateTime to TIMESTAMPTZ columns, which drifts whenever the database session timezone is not UTC. All timestamp binds across R2dbcSessionStore, R2dbcMessageStore, R2dbcActionStore, R2dbcAttachmentStore, and R2dbcModelConfigStore now go through dialect.bindInstant (Instant, and Instant-typed nulls), matching the platform convention; BO wall-clock fields stay UTC-naive and reads normalize via DatabaseInstant. The same pattern remains in R2dbcModelProviderStore (owned by another theater). Regression coverage: a scripted in-memory DatabaseClient models READ COMMITTED row locking and unique-key conflicts, with tests for tombstone revive, concurrent-update merge without lost writes, the delete-vs-upload TOCTOU, guard ordering, Instant binds, and non-UTC offset decode normalization. dc3-db-agentic gains its first test suite (26 tests); dc3-common-agentic keeps 206 green. Co-Authored-By: Claude Code --- .../repository/ReactiveAttachmentStore.java | 15 +- .../repository/ReactiveSessionStore.java | 24 +- .../service/impl/AttachmentFileCleaner.java | 58 +- .../service/impl/SessionServiceImpl.java | 22 +- .../impl/AttachmentFileCleanerTest.java | 120 +++ .../service/impl/SessionServiceImplTest.java | 64 +- dc3-db/dc3-db-agentic/pom.xml | 10 + .../pnoker/db/agentic/R2dbcActionStore.java | 42 +- .../db/agentic/R2dbcAttachmentStore.java | 59 +- .../pnoker/db/agentic/R2dbcMessageStore.java | 17 +- .../db/agentic/R2dbcModelConfigStore.java | 36 +- .../pnoker/db/agentic/R2dbcSessionStore.java | 220 ++-- .../db/agentic/AgenticStoreLayeringTest.java | 56 + .../db/agentic/R2dbcActionStoreTest.java | 118 +++ .../db/agentic/R2dbcAttachmentStoreTest.java | 184 ++++ .../db/agentic/R2dbcMessageStoreTest.java | 100 ++ .../db/agentic/R2dbcModelConfigStoreTest.java | 73 ++ .../db/agentic/R2dbcSessionStoreTest.java | 270 +++++ .../db/agentic/ScriptedAgenticDatabase.java | 975 ++++++++++++++++++ dc3/bin/check_r2dbc_migration.py | 5 + 20 files changed, 2303 insertions(+), 165 deletions(-) create mode 100644 dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleanerTest.java create mode 100644 dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/AgenticStoreLayeringTest.java create mode 100644 dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcActionStoreTest.java create mode 100644 dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcAttachmentStoreTest.java create mode 100644 dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcMessageStoreTest.java create mode 100644 dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcModelConfigStoreTest.java create mode 100644 dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcSessionStoreTest.java create mode 100644 dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/ScriptedAgenticDatabase.java 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 12b875ba5..3b2a94b6e 100644 --- a/dc3-db/dc3-db-agentic/pom.xml +++ b/dc3-db/dc3-db-agentic/pom.xml @@ -44,5 +44,15 @@ io.github.pnoker dc3-common-agentic + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + 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")