diff --git a/.gitignore b/.gitignore index 33c4e76..e08ba35 100644 --- a/.gitignore +++ b/.gitignore @@ -104,3 +104,7 @@ optd-sidecar/target/ *.iml .local-admin-credentials .local-mcp-token + +# Python bytecode from scripts/ +__pycache__/ +*.pyc diff --git a/CLAUDE.md b/CLAUDE.md index cc9df5b..cbe5242 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -296,6 +296,16 @@ broken. Assert the *outcome*, never the attempt: - **Mocks hide SDK breaks.** `tests/tools/test_mcp_structured_content.py` uses a `_FakeCallToolResult` with a hardcoded `.isError`, so it kept passing precisely when the real SDK stopped matching. Pin the dependency; a fake cannot catch this. +- **A self-host verification script must reach the DB the way the install does.** + `seed-review-suggestions.py` / `e2e-review-approvals.py` hardcoded + `sudo -u postgres psql`, which only exists on a bare-metal install — on the Compose + deployment `install.sh` actually produces, the documented verify command died before + testing anything. Both now resolve the path through `scripts/self-host/vaultdb.py`. +- **A test that mutates shared state must restore it, and only what it created.** The + same e2e suite parked every real `CODE_DERIVED` row by rewriting `source` to `USER` + and never restored it, so a run against a live install silently relabelled the user's + approved docs. It now copies rows to a scratch table and restores them, and its + cleanup deletes the planted row only while nothing references it. - **Never claim a check you did not run.** `install.sh` reported "up to date" when it could not reach npm; it now says it could not check. - **`set -e` + `read` at EOF aborts silently.** Prompts in `install.sh` use @@ -384,6 +394,53 @@ it against a real database — not a theoretical hardening pass. `POST /users/admin/reset` on every install that had run `setup-agent.sh`, since that mints an admin MCP token on each run. +- **An `Optional`-returning derived finder is an assertion that the key is unique.** + Spring Data throws `IncorrectResultSizeDataAccessException` ("Query did not return a + unique result: N results were returned") the moment it is not, and the row that broke + it never repairs itself, so the failure is permanent rather than transient. + `schema_documentation` had no unique constraint on + `(connection_id, object_type, object_name, parent_object, source)` and + `CodeSuggestionApplier.approve` took no row lock, so one bulk approve submitted twice + concurrently wrote 219 duplicate pairs. Every later approve touching one of those keys + threw, `CodeScanService.bulkDecide` swallowed it per item, and the Review queue + reported "Approved 0 of 2" — with all 198 pending SCHEMA_DOC suggestions wedged. + Three-part fix, and all three are load-bearing: + 1. `V116__dedupe_schema_documentation.sql` + `SchemaDocumentationDedupeInitializer` + (no Flyway here, so the initializer is what actually applies it) collapse the + duplicates and add `ux_schema_doc_target`, keyed on + `coalesce(parent_object,'')` because Postgres treats NULLs as distinct. + 2. Those finders now return `List`, and `SchemaDocumentationDeduplicator.collapse` + keeps the newest row, repoints any `applied_doc_id` off the rows it deletes, and + drops their RAG embeddings. Do not restore an `Optional` variant — legacy installs + still carry duplicates until the initializer runs. + 3. `approve`/`reject` load the suggestion via `findByIdForUpdate` (`PESSIMISTIC_WRITE`) + so the concurrent double-submit that created the duplicates blocks instead of racing. +- **`applied_doc_id` is a loose reference, not an FK.** Deleting a `schema_documentation` + row it points at raises nothing and dangles silently — repoint before deleting. +- **Approve *updates* the row an earlier scan wrote**, so a "freshly approved" doc row + carries a historical `created_at`. A test that plants an "old" duplicate with a + hardcoded past date can easily plant the *newer* of the two and assert nothing; anchor + fixture timestamps to the real row's `created_at`. +- **A write's blast radius decides what to invalidate, not the endpoint you called.** + Approving a code-scan suggestion writes `code_knowledge_suggestion` *and* + `schema_documentation` (served by `brain/notes`, which backs the Write-notes tab + and its coverage counts) *and* `rag_documents` *and*, for KNOWLEDGE_ENTRY, a + company knowledge entry. The decide hooks invalidated only `codeScan` + + `companyKnowledge`, so every schema-doc-derived count stayed stale until the user + reloaded the page. `invalidateAfterDecision` in `useCodeScan.js` is the single + place that lists them; add to it when an approval starts writing something new. +- **`@PreUpdate` does not fire on insert, so `updatedAt` is null on a brand-new row.** + Sorting "newest first" on `updatedAt` alone with nulls last therefore sends every + freshly created row to the *bottom* — which is why a just-approved note did not + appear at the top of the Write-notes list. Sort on + `COALESCE(updatedAt, createdAt)` (`BrainNoteService.touchedAt`, + `CompanyKnowledgeEntryRepository.findByConnectionIdOrderByRecency`). +- **Suggestion list order depends on the status being viewed.** PENDING is a work + queue → `confidence DESC`. APPROVED/REJECTED are history → `decidedAt DESC NULLS + LAST` so the decision you just made is at the top; confidence-sorting a decided + list scattered fresh approvals among hundreds of older ones + (`CodeScanService.sortFor`). + ### Endpoint Authorization Rules - **Authentication is not authorization.** `SecurityConfig` only asserts diff --git a/backend/src/main/java/com/dbaagent/config/SchemaDocumentationDedupeInitializer.java b/backend/src/main/java/com/dbaagent/config/SchemaDocumentationDedupeInitializer.java new file mode 100644 index 0000000..5e7c962 --- /dev/null +++ b/backend/src/main/java/com/dbaagent/config/SchemaDocumentationDedupeInitializer.java @@ -0,0 +1,116 @@ +package com.dbaagent.config; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.DependsOn; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.transaction.PlatformTransactionManager; + +import javax.sql.DataSource; + +/** + * Applies {@code V116__dedupe_schema_documentation.sql} at startup: collapses + * duplicate {@code schema_documentation} rows and adds the unique index on the + * logical key. + * + *

This repo has no Flyway runtime — {@code db/migration} is a hand-maintained + * changelog and Hibernate {@code ddl-auto=update} never adds an index the entity + * does not declare. Without this, self-host installs carrying duplicates from a + * double-submitted bulk approve stay wedged: every SCHEMA_DOC approve throws + * {@code Query did not return a unique result}. Mirrors + * {@link SchemaDocumentationSourceCompatibilityInitializer}. + * + *

Idempotent and cheap on a clean install: the index exists, so it returns + * before touching a row. + */ +@Configuration +@Slf4j +public class SchemaDocumentationDedupeInitializer { + + private static final String TABLE = "schema_documentation"; + private static final String INDEX = "ux_schema_doc_target"; + + @Bean("schemaDocumentationDedupeBootstrap") + @DependsOn("entityManagerFactory") + public Object schemaDocumentationDedupeBootstrap(DataSource dataSource, + PlatformTransactionManager txManager) { + JdbcTemplate jdbc = new JdbcTemplate(dataSource); + if (!tableExists(jdbc, TABLE)) { + return new Object(); + } + if (indexExists(jdbc, INDEX)) { + return new Object(); + } + + // One transaction: a half-applied dedupe (rows deleted, index missing) + // would silently re-accumulate duplicates until the next boot. + new TransactionTemplate(txManager).executeWithoutResult(status -> { + int repointed = jdbc.update(""" + UPDATE code_knowledge_suggestion s + SET applied_doc_id = l.keep_id + FROM (%s) l + WHERE s.applied_doc_id = l.id + """.formatted(LOSERS)); + + int embeddings = jdbc.update( + "DELETE FROM rag_documents WHERE id IN (SELECT id FROM (%s) l)".formatted(LOSERS)); + + int removed = jdbc.update( + "DELETE FROM schema_documentation WHERE id IN (SELECT id FROM (%s) l)".formatted(LOSERS)); + + jdbc.execute(""" + CREATE UNIQUE INDEX IF NOT EXISTS %s + ON %s (connection_id, object_type, object_name, coalesce(parent_object, ''), source) + """.formatted(INDEX, TABLE)); + + if (removed > 0) { + log.warn("Deduped {}: removed {} duplicate rows, {} orphaned embeddings, " + + "repointed {} applied_doc_id references", + TABLE, removed, embeddings, repointed); + } + log.info("Ensured unique index {} on {}", INDEX, TABLE); + }); + return new Object(); + } + + /** + * Every row but the newest within each logical key. Newest wins because it is + * the row existing {@code applied_doc_id} references point at; {@code id} + * breaks ties for rows written in the same clock tick. {@code coalesce} on + * {@code parent_object} because Postgres treats NULLs as distinct, so TABLE + * rows would otherwise never group together. + */ + private static final String LOSERS = """ + SELECT id, keep_id FROM ( + SELECT id, + first_value(id) OVER w AS keep_id, + row_number() OVER w AS rn + FROM schema_documentation + WINDOW w AS ( + PARTITION BY connection_id, object_type, object_name, + coalesce(parent_object, ''), source + ORDER BY created_at DESC NULLS LAST, id DESC + ) + ) ranked WHERE rn > 1 + """; + + private boolean tableExists(JdbcTemplate jdbc, String tableName) { + Integer count = jdbc.queryForObject(""" + SELECT COUNT(*) + FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = ? + """, Integer.class, tableName); + return count != null && count > 0; + } + + private boolean indexExists(JdbcTemplate jdbc, String indexName) { + Integer count = jdbc.queryForObject(""" + SELECT COUNT(*) + FROM pg_indexes + WHERE schemaname = 'public' AND indexname = ? + """, Integer.class, indexName); + return count != null && count > 0; + } +} diff --git a/backend/src/main/java/com/dbaagent/repository/CodeKnowledgeSuggestionRepository.java b/backend/src/main/java/com/dbaagent/repository/CodeKnowledgeSuggestionRepository.java index f75bf44..4111315 100644 --- a/backend/src/main/java/com/dbaagent/repository/CodeKnowledgeSuggestionRepository.java +++ b/backend/src/main/java/com/dbaagent/repository/CodeKnowledgeSuggestionRepository.java @@ -3,14 +3,44 @@ import com.dbaagent.model.code.CodeKnowledgeSuggestion; import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; +import jakarta.persistence.LockModeType; import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Lock; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; import org.springframework.stereotype.Repository; +import java.util.Collection; import java.util.List; +import java.util.Optional; @Repository public interface CodeKnowledgeSuggestionRepository extends JpaRepository { + /** + * Row-locking load used by approve/reject. Without it two concurrent bulk + * decides both read the same suggestion as PENDING and both materialize a + * {@code schema_documentation} row — the duplicate-row bug that + * {@code V116__dedupe_schema_documentation.sql} had to clean up. The second + * caller now blocks, then sees APPROVED and returns early. + */ + @Lock(LockModeType.PESSIMISTIC_WRITE) + @Query("SELECT s FROM CodeKnowledgeSuggestion s WHERE s.id = :id") + Optional findByIdForUpdate(@Param("id") String id); + + /** + * Repoints approvals at the surviving row when duplicate schema_documentation + * rows are collapsed. {@code applied_doc_id} is a loose reference, not an FK, + * so deleting a duplicate would otherwise leave a suggestion pointing at a row + * that no longer exists — silently, since nothing enforces it. + */ + @Modifying(flushAutomatically = true) + @Query("UPDATE CodeKnowledgeSuggestion s SET s.appliedDocId = :keepId " + + "WHERE s.appliedDocId IN :staleIds") + int repointAppliedDocId(@Param("keepId") String keepId, + @Param("staleIds") Collection staleIds); + Page findByConnectionIdAndStatus( String connectionId, CodeKnowledgeSuggestion.Status status, diff --git a/backend/src/main/java/com/dbaagent/repository/SchemaDocumentationRepository.java b/backend/src/main/java/com/dbaagent/repository/SchemaDocumentationRepository.java index bd23af1..1ed99e8 100644 --- a/backend/src/main/java/com/dbaagent/repository/SchemaDocumentationRepository.java +++ b/backend/src/main/java/com/dbaagent/repository/SchemaDocumentationRepository.java @@ -21,7 +21,15 @@ List findByConnectionIdAndObjectType( SchemaDocumentation.DocumentationType objectType ); - Optional findByConnectionIdAndObjectTypeAndObjectName( + /** + * Returns a {@link List}, never an {@link Optional} — the logical key is not + * unique in data written before {@code V116__dedupe_schema_documentation.sql} + * added the constraint, and an {@code Optional} finder throws + * {@code IncorrectResultSizeDataAccessException} on a legacy duplicate rather + * than letting the caller repair it. Collapse matches with + * {@link com.dbaagent.service.SchemaDocumentationDeduplicator}. + */ + List findByConnectionIdAndObjectTypeAndObjectName( String connectionId, SchemaDocumentation.DocumentationType objectType, String objectName @@ -51,12 +59,13 @@ AND TRIM(d.businessTerms) <> '' """) long countWithBusinessTerms(@Param("connectionId") String connectionId); - // Upsert support: find existing AI doc to update instead of creating duplicates - Optional findByConnectionIdAndObjectTypeAndObjectNameAndSource( + // Upsert support: find existing doc to update instead of creating duplicates. + // List-returning for the same reason as findByConnectionIdAndObjectTypeAndObjectName above. + List findByConnectionIdAndObjectTypeAndObjectNameAndSource( String connectionId, SchemaDocumentation.DocumentationType objectType, String objectName, DocumentationSource source); - Optional findByConnectionIdAndObjectTypeAndObjectNameAndParentObjectAndSource( + List findByConnectionIdAndObjectTypeAndObjectNameAndParentObjectAndSource( String connectionId, SchemaDocumentation.DocumentationType objectType, String objectName, String parentObject, DocumentationSource source); diff --git a/backend/src/main/java/com/dbaagent/service/SchemaDescriptionService.java b/backend/src/main/java/com/dbaagent/service/SchemaDescriptionService.java index 19d82f8..7b32756 100644 --- a/backend/src/main/java/com/dbaagent/service/SchemaDescriptionService.java +++ b/backend/src/main/java/com/dbaagent/service/SchemaDescriptionService.java @@ -34,6 +34,7 @@ public class SchemaDescriptionService { private final ColumnProfileRepository columnProfileRepo; private final InferredTableRelationshipRepository inferredRelationshipRepository; private final TrainingService trainingService; + private final SchemaDocumentationDeduplicator schemaDocDeduplicator; private final ConnectionService connectionService; private final DatabaseProviderRegistry providerRegistry; private final ObjectMapper objectMapper = new ObjectMapper(); @@ -71,6 +72,7 @@ public SchemaDescriptionService( ColumnProfileRepository columnProfileRepo, InferredTableRelationshipRepository inferredRelationshipRepository, TrainingService trainingService, + SchemaDocumentationDeduplicator schemaDocDeduplicator, ConnectionService connectionService, DatabaseProviderRegistry providerRegistry, @Value("${brain.description.ai-concurrency:4}") int aiConcurrency) { @@ -80,6 +82,7 @@ public SchemaDescriptionService( this.columnProfileRepo = columnProfileRepo; this.inferredRelationshipRepository = inferredRelationshipRepository; this.trainingService = trainingService; + this.schemaDocDeduplicator = schemaDocDeduplicator; this.connectionService = connectionService; this.providerRegistry = providerRegistry; this.aiConcurrency = Math.max(1, aiConcurrency); @@ -417,13 +420,14 @@ private int saveTableDescription(String connectionId, TableDescription desc, : desc.getTableName(); // Upsert table-level doc (find existing AI doc or create new) - var existingTableDoc = schemaDocRepo - .findByConnectionIdAndObjectTypeAndObjectNameAndSource( + var existingTableDoc = schemaDocDeduplicator.collapse( + schemaDocRepo.findByConnectionIdAndObjectTypeAndObjectNameAndSource( connectionId, DocumentationType.TABLE, objectName, - DocumentationSource.AI_GENERATED); + DocumentationSource.AI_GENERATED), + objectName + " (TABLE, AI_GENERATED)"); SchemaDocumentation tableDoc; - if (existingTableDoc.isPresent()) { - tableDoc = existingTableDoc.get(); + if (existingTableDoc != null) { + tableDoc = existingTableDoc; tableDoc.setDescription(desc.getTableDescription()); tableDoc.setBusinessTerms(desc.getBusinessTerms()); tableDoc.setConfidence(desc.getConfidence()); @@ -445,13 +449,14 @@ private int saveTableDescription(String connectionId, TableDescription desc, // Upsert column-level docs for (var col : desc.getColumns()) { if (col.getDescription() == null || col.getDescription().isBlank()) continue; - var existingColDoc = schemaDocRepo - .findByConnectionIdAndObjectTypeAndObjectNameAndParentObjectAndSource( + var existingColDoc = schemaDocDeduplicator.collapse( + schemaDocRepo.findByConnectionIdAndObjectTypeAndObjectNameAndParentObjectAndSource( connectionId, DocumentationType.COLUMN, col.getName(), - objectName, DocumentationSource.AI_GENERATED); + objectName, DocumentationSource.AI_GENERATED), + objectName + "." + col.getName() + " (COLUMN, AI_GENERATED)"); SchemaDocumentation colDoc; - if (existingColDoc.isPresent()) { - colDoc = existingColDoc.get(); + if (existingColDoc != null) { + colDoc = existingColDoc; colDoc.setDescription(col.getDescription()); colDoc.setConfidence(col.getConfidence()); } else { diff --git a/backend/src/main/java/com/dbaagent/service/SchemaDocumentationDeduplicator.java b/backend/src/main/java/com/dbaagent/service/SchemaDocumentationDeduplicator.java new file mode 100644 index 0000000..b34a2d4 --- /dev/null +++ b/backend/src/main/java/com/dbaagent/service/SchemaDocumentationDeduplicator.java @@ -0,0 +1,93 @@ +package com.dbaagent.service; + +import com.dbaagent.model.SchemaDocumentation; +import com.dbaagent.repository.CodeKnowledgeSuggestionRepository; +import com.dbaagent.repository.SchemaDocumentationRepository; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +import java.time.LocalDateTime; +import java.util.Comparator; +import java.util.List; + +/** + * Collapses duplicate {@link SchemaDocumentation} rows that share a logical key. + * + *

{@code schema_documentation} carried no unique constraint on + * {@code (connection_id, object_type, object_name, parent_object, source)} until + * {@code V116__dedupe_schema_documentation.sql}, and + * {@link com.dbaagent.service.codescan.CodeSuggestionApplier#approve} took no row + * lock — so a bulk approve submitted twice concurrently wrote two identical rows + * per suggestion. Every later upsert against that key then threw + * {@code IncorrectResultSizeDataAccessException: Query did not return a unique + * result}, which {@code CodeScanService.bulkDecide} swallowed into + * "Approved 0 of N". The duplicate never self-heals, so the suggestion stays + * permanently unapprovable. + * + *

Repairing on read is what unwedges installs whose duplicates predate the + * constraint: keep the newest row (the one existing {@code applied_doc_id} + * references), drop the rest along with their RAG embeddings. + */ +@Component +@RequiredArgsConstructor +@Slf4j +public class SchemaDocumentationDeduplicator { + + private final SchemaDocumentationRepository schemaDocRepository; + private final CodeKnowledgeSuggestionRepository suggestionRepository; + private final TrainingService trainingService; + + /** + * Reduces {@code matches} to at most one row, deleting any extras. + * + * @return the surviving row, or {@code null} when {@code matches} is empty + * (caller creates a fresh row). + */ + @Transactional + public SchemaDocumentation collapse(List matches, String context) { + if (matches == null || matches.isEmpty()) { + return null; + } + if (matches.size() == 1) { + return matches.get(0); + } + + // Newest wins: it is the row a prior approval linked via applied_doc_id, + // so keeping it preserves those references. id breaks ties on rows + // written inside the same clock tick. + List ordered = matches.stream() + .sorted(Comparator + .comparing(SchemaDocumentation::getCreatedAt, + Comparator.nullsFirst(Comparator.naturalOrder())) + .thenComparing(SchemaDocumentation::getId, + Comparator.nullsFirst(Comparator.naturalOrder()))) + .toList(); + + SchemaDocumentation survivor = ordered.get(ordered.size() - 1); + List stale = ordered.subList(0, ordered.size() - 1); + log.warn("Collapsing {} duplicate schema_documentation rows for {} — keeping {}", + ordered.size(), context, survivor.getId()); + + // Which row wins is a heuristic; leaving another approval pointing at a + // deleted id is not acceptable either way, so repoint before deleting. + int repointed = suggestionRepository.repointAppliedDocId( + survivor.getId(), stale.stream().map(SchemaDocumentation::getId).toList()); + if (repointed > 0) { + log.info("Repointed {} applied_doc_id reference(s) to {}", repointed, survivor.getId()); + } + + for (SchemaDocumentation doc : stale) { + try { + trainingService.deleteDocumentationEmbedding(doc.getConnectionId(), doc.getId()); + } catch (Exception e) { + // A stranded embedding degrades retrieval; a failed delete must not + // block the approval the caller is in the middle of. + log.warn("Failed to delete embedding for duplicate doc {}: {}", doc.getId(), e.getMessage()); + } + schemaDocRepository.delete(doc); + } + return survivor; + } +} diff --git a/backend/src/main/java/com/dbaagent/service/SchemaDriftListener.java b/backend/src/main/java/com/dbaagent/service/SchemaDriftListener.java index 29b33fd..b86ef9e 100644 --- a/backend/src/main/java/com/dbaagent/service/SchemaDriftListener.java +++ b/backend/src/main/java/com/dbaagent/service/SchemaDriftListener.java @@ -78,13 +78,14 @@ public void onTablesRemoved(String connectionId, List sampleRemovedTable // Delete table-level doc (if AI-generated). Now also cascade // the embedding so the RAG store doesn't keep returning hits // for a table that no longer exists. + // Iterate: the logical key was not unique before V116, so a legacy + // install can hold more than one row here and an Optional finder + // would throw instead of cleaning either of them up. schemaDocRepo.findByConnectionIdAndObjectTypeAndObjectName( connectionId, SchemaDocumentation.DocumentationType.TABLE, canonical) - .ifPresent(doc -> { - if (doc.getSource() == DocumentationSource.AI_GENERATED) { - deleteWithEmbedding(doc, "dropped table " + canonical); - } - }); + .stream() + .filter(doc -> doc.getSource() == DocumentationSource.AI_GENERATED) + .forEach(doc -> deleteWithEmbedding(doc, "dropped table " + canonical)); // Delete column-level docs for this table (if AI-generated) var columnDocs = schemaDocRepo.findByConnectionIdAndParentObject(connectionId, canonical); diff --git a/backend/src/main/java/com/dbaagent/service/brain/core/BrainNoteService.java b/backend/src/main/java/com/dbaagent/service/brain/core/BrainNoteService.java index edac370..2680f25 100644 --- a/backend/src/main/java/com/dbaagent/service/brain/core/BrainNoteService.java +++ b/backend/src/main/java/com/dbaagent/service/brain/core/BrainNoteService.java @@ -34,12 +34,27 @@ public List getNotes(String connectionId, String scopeType, S List docs = schemaDocumentationRepository.findByConnectionId(connectionId); return docs.stream() .filter(doc -> matchesScope(doc, scopeType, tableName, columnName)) - .sorted(Comparator.comparing(SchemaDocumentation::getUpdatedAt, + .sorted(Comparator.comparing(BrainNoteService::touchedAt, Comparator.nullsLast(Comparator.reverseOrder()))) .map(doc -> toResponse(doc, lastSnapshotAt)) .collect(Collectors.toList()); } + /** + * Recency for sorting: {@code updatedAt} falls back to {@code createdAt}. + * + *

{@code @PreUpdate} never fires on an insert, so a note that has only ever + * been created has a null {@code updatedAt}. Sorting on {@code updatedAt} alone + * with nulls last therefore pushed every brand-new note to the *bottom* — the + * opposite of what "newest first" means, and why a freshly approved code-scan + * suggestion did not show up at the top of the list. Same + * {@code COALESCE(updatedAt, createdAt)} convention as + * {@code CompanyKnowledgeEntryRepository.findByConnectionIdOrderByRecency}. + */ + private static LocalDateTime touchedAt(SchemaDocumentation doc) { + return doc.getUpdatedAt() != null ? doc.getUpdatedAt() : doc.getCreatedAt(); + } + public BrainNoteResponse createNote(BrainNoteRequest request) { validateRequest(request, true); SchemaDocumentation doc = SchemaDocumentation.builder() diff --git a/backend/src/main/java/com/dbaagent/service/codescan/CodeScanService.java b/backend/src/main/java/com/dbaagent/service/codescan/CodeScanService.java index e9e0513..5b4a6da 100644 --- a/backend/src/main/java/com/dbaagent/service/codescan/CodeScanService.java +++ b/backend/src/main/java/com/dbaagent/service/codescan/CodeScanService.java @@ -553,8 +553,30 @@ public Page listSuggestions(String connectionId, CodeKnowledgeSuggestion.Status status, int page, int size) { - var pageable = PageRequest.of(page, size, Sort.by(Sort.Direction.DESC, "confidence", "createdAt")); - return suggestionRepository.findByConnectionIdAndStatus(connectionId, status, pageable); + return suggestionRepository.findByConnectionIdAndStatus( + connectionId, status, PageRequest.of(page, size, sortFor(status))); + } + + /** + * Sort depends on what the reviewer is looking at. + * + *

PENDING is a work queue — highest confidence first, so the best candidates + * are the ones you see. Anything already decided is a history view, and the + * question there is "what did I just do", so it leads with the most recent + * decision. Sorting decided rows by confidence scattered a fresh approval + * somewhere in the middle of hundreds of older ones. + * + *

{@code decidedAt} is null on rows decided before it was recorded, hence + * nullsLast with a createdAt fallback. + */ + private static Sort sortFor(CodeKnowledgeSuggestion.Status status) { + if (status == CodeKnowledgeSuggestion.Status.PENDING) { + return Sort.by(Sort.Direction.DESC, "confidence", "createdAt"); + } + return Sort.by( + new Sort.Order(Sort.Direction.DESC, "decidedAt", Sort.NullHandling.NULLS_LAST), + new Sort.Order(Sort.Direction.DESC, "createdAt") + ); } /** diff --git a/backend/src/main/java/com/dbaagent/service/codescan/CodeSuggestionApplier.java b/backend/src/main/java/com/dbaagent/service/codescan/CodeSuggestionApplier.java index 174bf64..b77a781 100644 --- a/backend/src/main/java/com/dbaagent/service/codescan/CodeSuggestionApplier.java +++ b/backend/src/main/java/com/dbaagent/service/codescan/CodeSuggestionApplier.java @@ -7,6 +7,7 @@ import com.dbaagent.repository.CodeKnowledgeSuggestionRepository; import com.dbaagent.repository.SchemaDocumentationRepository; import com.dbaagent.service.CompanyKnowledgeService; +import com.dbaagent.service.SchemaDocumentationDeduplicator; import com.dbaagent.service.SchemaScannerService; import com.dbaagent.service.TrainingService; import lombok.RequiredArgsConstructor; @@ -17,7 +18,6 @@ import java.time.LocalDateTime; import java.util.List; import java.util.Map; -import java.util.Optional; /** * Materializes an APPROVED suggestion into a real @@ -35,10 +35,11 @@ public class CodeSuggestionApplier { private final CompanyKnowledgeService companyKnowledgeService; private final TrainingService trainingService; private final SchemaScannerService schemaScannerService; + private final SchemaDocumentationDeduplicator schemaDocDeduplicator; @Transactional public CodeKnowledgeSuggestion approve(String suggestionId, String decidedBy, String note) { - CodeKnowledgeSuggestion suggestion = suggestionRepository.findById(suggestionId) + CodeKnowledgeSuggestion suggestion = suggestionRepository.findByIdForUpdate(suggestionId) .orElseThrow(() -> new IllegalArgumentException("Suggestion not found: " + suggestionId)); if (suggestion.getStatus() == CodeKnowledgeSuggestion.Status.APPROVED) { return suggestion; @@ -62,7 +63,7 @@ public CodeKnowledgeSuggestion approve(String suggestionId, String decidedBy, St @Transactional public CodeKnowledgeSuggestion reject(String suggestionId, String decidedBy, String note) { - CodeKnowledgeSuggestion suggestion = suggestionRepository.findById(suggestionId) + CodeKnowledgeSuggestion suggestion = suggestionRepository.findByIdForUpdate(suggestionId) .orElseThrow(() -> new IllegalArgumentException("Suggestion not found: " + suggestionId)); if (suggestion.getStatus() == CodeKnowledgeSuggestion.Status.REJECTED) { return suggestion; @@ -112,11 +113,14 @@ private void applySchemaDoc(CodeKnowledgeSuggestion suggestion, String decidedBy } // Scope the upsert to our own CODE_DERIVED rows AND, for columns, - // include parent_object — same column name can exist in many tables, - // so a (connection, COLUMN, "status", CODE_DERIVED) lookup would - // otherwise return multiple rows and throw "Query did not return a - // unique result". - Optional existing; + // include parent_object — the same column name exists in many tables, so a + // (connection, COLUMN, "status", CODE_DERIVED) lookup would match all of them. + // + // Even with the full key this can match more than one row on installs whose + // duplicates predate V116's unique index, so collapse rather than assume one: + // an Optional finder here threw "Query did not return a unique result" and + // wedged the suggestion forever. + List existing; if (objectType == SchemaDocumentation.DocumentationType.COLUMN) { existing = schemaDocRepository .findByConnectionIdAndObjectTypeAndObjectNameAndParentObjectAndSource( @@ -136,7 +140,12 @@ private void applySchemaDoc(CodeKnowledgeSuggestion suggestion, String decidedBy ); } - SchemaDocumentation doc = existing.orElseGet(SchemaDocumentation::new); + String docKey = (parentObject == null ? objectName : parentObject + "." + objectName) + + " (" + objectType + ", CODE_DERIVED)"; + SchemaDocumentation doc = schemaDocDeduplicator.collapse(existing, docKey); + if (doc == null) { + doc = new SchemaDocumentation(); + } doc.setConnectionId(suggestion.getConnectionId()); doc.setObjectType(objectType); doc.setObjectName(objectName); diff --git a/backend/src/main/resources/db/migration/V116__dedupe_schema_documentation.sql b/backend/src/main/resources/db/migration/V116__dedupe_schema_documentation.sql new file mode 100644 index 0000000..4fd28b5 --- /dev/null +++ b/backend/src/main/resources/db/migration/V116__dedupe_schema_documentation.sql @@ -0,0 +1,53 @@ +-- Collapse duplicate schema_documentation rows and make the logical key unique. +-- +-- schema_documentation never carried a unique constraint on +-- (connection_id, object_type, object_name, parent_object, source), and +-- CodeSuggestionApplier.approve took no row lock — so a bulk approve submitted +-- twice concurrently wrote two identical rows per suggestion. Every later upsert +-- against such a key then threw +-- IncorrectResultSizeDataAccessException: Query did not return a unique result +-- which CodeScanService.bulkDecide swallowed into "Approved 0 of N", leaving the +-- whole Review queue permanently unapprovable. +-- +-- NOTE: this repo has no Flyway runtime. Applied at startup, idempotently, by +-- SchemaDocumentationDedupeInitializer — this file is the changelog of record. + +BEGIN; + +CREATE TEMP TABLE schema_doc_dupe_losers ON COMMIT DROP AS +SELECT id, keep_id +FROM ( + SELECT id, + first_value(id) OVER w AS keep_id, + row_number() OVER w AS rn + FROM schema_documentation + WINDOW w AS ( + PARTITION BY connection_id, object_type, object_name, + coalesce(parent_object, ''), source + ORDER BY created_at DESC NULLS LAST, id DESC + ) +) ranked +WHERE rn > 1; + +-- Keep the newest row: it is the one existing applied_doc_id references point at. +-- Repoint any that reference a loser before it disappears (applied_doc_id is a +-- loose reference, not a real FK, so a stale value would fail silently). +UPDATE code_knowledge_suggestion s +SET applied_doc_id = l.keep_id +FROM schema_doc_dupe_losers l +WHERE s.applied_doc_id = l.id; + +-- RAG embeddings for documentation are keyed by the doc id, so the loser's +-- vector must go with it or retrieval keeps returning the orphan. +DELETE FROM rag_documents WHERE id IN (SELECT id FROM schema_doc_dupe_losers); + +DELETE FROM schema_documentation WHERE id IN (SELECT id FROM schema_doc_dupe_losers); + +-- coalesce(parent_object, '') because Postgres treats NULLs as distinct: without +-- it, TABLE rows (parent_object IS NULL) would never collide. +CREATE UNIQUE INDEX IF NOT EXISTS ux_schema_doc_target + ON schema_documentation ( + connection_id, object_type, object_name, coalesce(parent_object, ''), source + ); + +COMMIT; diff --git a/backend/src/test/java/com/dbaagent/service/SchemaDescriptionServiceTest.java b/backend/src/test/java/com/dbaagent/service/SchemaDescriptionServiceTest.java index 47028b9..b6b07c1 100644 --- a/backend/src/test/java/com/dbaagent/service/SchemaDescriptionServiceTest.java +++ b/backend/src/test/java/com/dbaagent/service/SchemaDescriptionServiceTest.java @@ -30,6 +30,7 @@ class SchemaDescriptionServiceTest { @Mock private com.dbaagent.repository.ColumnProfileRepository columnProfileRepo; @Mock private com.dbaagent.repository.InferredTableRelationshipRepository inferredRelationshipRepository; @Mock private TrainingService trainingService; + @Mock private com.dbaagent.repository.CodeKnowledgeSuggestionRepository codeSuggestionRepo; @Mock private ConnectionService connectionService; @Mock private com.dbaagent.provider.DatabaseProviderRegistry providerRegistry; @@ -42,8 +43,9 @@ void setUp() { .thenReturn(List.of()); service = new SchemaDescriptionService( chatClientBuilder, schemaScannerService, schemaDocRepo, - columnProfileRepo, inferredRelationshipRepository, trainingService, connectionService, - providerRegistry, 4 + columnProfileRepo, inferredRelationshipRepository, trainingService, + new SchemaDocumentationDeduplicator(schemaDocRepo, codeSuggestionRepo, trainingService), + connectionService, providerRegistry, 4 ); } @@ -173,7 +175,7 @@ void forceRegenerateOverridesDeltaOnlyModeAndRegeneratesExistingAiDocs() throws .build() )); when(schemaDocRepo.findByConnectionIdAndObjectTypeAndObjectNameAndSource(any(), any(), any(), any())) - .thenReturn(Optional.empty()); + .thenReturn(List.of()); when(schemaDocRepo.save(any())).thenAnswer(i -> i.getArgument(0)); when(connectionService.isDataSamplingEnabled("conn1")).thenReturn(false); var schema = buildSchemaWithTables("users"); diff --git a/backend/src/test/java/com/dbaagent/service/SchemaDriftListenerColumnsTest.java b/backend/src/test/java/com/dbaagent/service/SchemaDriftListenerColumnsTest.java index bf7faab..94a2b00 100644 --- a/backend/src/test/java/com/dbaagent/service/SchemaDriftListenerColumnsTest.java +++ b/backend/src/test/java/com/dbaagent/service/SchemaDriftListenerColumnsTest.java @@ -94,7 +94,7 @@ void onTablesRemovedNowCleansEmbeddings() { tableDoc.setObjectType(SchemaDocumentation.DocumentationType.TABLE); when(repo.findByConnectionIdAndObjectTypeAndObjectName( "c1", SchemaDocumentation.DocumentationType.TABLE, "OLD_TABLE")) - .thenReturn(java.util.Optional.of(tableDoc)); + .thenReturn(List.of(tableDoc)); when(repo.findByConnectionIdAndParentObject("c1", "OLD_TABLE")) .thenReturn(List.of()); diff --git a/backend/src/test/java/com/dbaagent/service/brain/core/BrainNoteServiceOrderingTest.java b/backend/src/test/java/com/dbaagent/service/brain/core/BrainNoteServiceOrderingTest.java new file mode 100644 index 0000000..31c2891 --- /dev/null +++ b/backend/src/test/java/com/dbaagent/service/brain/core/BrainNoteServiceOrderingTest.java @@ -0,0 +1,98 @@ +package com.dbaagent.service.brain.core; + +import com.dbaagent.model.DocumentationSource; +import com.dbaagent.model.SchemaDocumentation; +import com.dbaagent.model.brain.BrainNoteResponse; +import com.dbaagent.repository.SchemaDocumentationRepository; +import com.dbaagent.repository.SchemaSnapshotRepository; +import com.dbaagent.service.TrainingService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.LocalDateTime; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.when; + +/** + * The Write-notes list is "newest first", and a note approved from the code-scan + * Review queue must land at the top of it. + */ +@ExtendWith(MockitoExtension.class) +class BrainNoteServiceOrderingTest { + + @Mock private SchemaDocumentationRepository schemaDocumentationRepository; + @Mock private SchemaSnapshotRepository schemaSnapshotRepository; + @Mock private TrainingService trainingService; + + private BrainNoteService service; + + @BeforeEach + void setUp() { + service = new BrainNoteService( + schemaDocumentationRepository, schemaSnapshotRepository, trainingService); + } + + /** + * Regression: {@code @PreUpdate} does not fire on insert, so a just-created note + * has a null {@code updatedAt}. Sorting on {@code updatedAt} with nulls last sent + * every brand-new note to the bottom of the list. + */ + @Test + void freshlyCreatedNoteSortsAboveOlderEditedOnes() { + SchemaDocumentation editedYesterday = doc( + "old_col", + LocalDateTime.of(2026, 8, 1, 9, 0), // created + LocalDateTime.of(2026, 8, 21, 9, 0)); // updated + SchemaDocumentation createdJustNow = doc( + "new_col", + LocalDateTime.of(2026, 8, 22, 17, 0), + null); // never edited + when(schemaDocumentationRepository.findByConnectionId("c1")) + .thenReturn(List.of(editedYesterday, createdJustNow)); + + List notes = service.getNotes("c1", null, null, null); + + assertThat(notes).extracting(BrainNoteResponse::getColumnName) + .containsExactly("new_col", "old_col"); + } + + @Test + void editedNoteSortsAboveNewerButUntouchedOne() { + SchemaDocumentation createdEarlier = doc( + "edited_col", + LocalDateTime.of(2026, 8, 1, 9, 0), + LocalDateTime.of(2026, 8, 22, 18, 0)); // edited most recently + SchemaDocumentation createdLater = doc( + "untouched_col", + LocalDateTime.of(2026, 8, 22, 17, 0), + null); + when(schemaDocumentationRepository.findByConnectionId("c1")) + .thenReturn(List.of(createdLater, createdEarlier)); + + List notes = service.getNotes("c1", null, null, null); + + assertThat(notes).extracting(BrainNoteResponse::getColumnName) + .containsExactly("edited_col", "untouched_col"); + } + + private static SchemaDocumentation doc(String columnName, + LocalDateTime createdAt, + LocalDateTime updatedAt) { + SchemaDocumentation doc = new SchemaDocumentation(); + doc.setId(columnName); + doc.setConnectionId("c1"); + doc.setObjectType(SchemaDocumentation.DocumentationType.COLUMN); + doc.setObjectName(columnName); + doc.setParentObject("analytics.orders"); + doc.setDescription("desc for " + columnName); + doc.setSource(DocumentationSource.CODE_DERIVED); + doc.setCreatedAt(createdAt); + doc.setUpdatedAt(updatedAt); + return doc; + } +} diff --git a/backend/src/test/java/com/dbaagent/service/codescan/CodeSuggestionApplierTest.java b/backend/src/test/java/com/dbaagent/service/codescan/CodeSuggestionApplierTest.java index 773806f..d42c208 100644 --- a/backend/src/test/java/com/dbaagent/service/codescan/CodeSuggestionApplierTest.java +++ b/backend/src/test/java/com/dbaagent/service/codescan/CodeSuggestionApplierTest.java @@ -7,6 +7,7 @@ import com.dbaagent.repository.CodeKnowledgeSuggestionRepository; import com.dbaagent.repository.SchemaDocumentationRepository; import com.dbaagent.service.CompanyKnowledgeService; +import com.dbaagent.service.SchemaDocumentationDeduplicator; import com.dbaagent.service.SchemaScannerService; import com.dbaagent.service.TrainingService; import org.junit.jupiter.api.BeforeEach; @@ -16,6 +17,7 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import java.time.LocalDateTime; import java.util.List; import java.util.Map; import java.util.Optional; @@ -24,6 +26,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -38,15 +41,20 @@ class CodeSuggestionApplierTest { @Mock private SchemaScannerService schemaScannerService; private CodeSuggestionApplier applier; + private SchemaDocumentationDeduplicator deduplicator; @BeforeEach void setUp() { + // Real deduplicator over the mocked repo: the collapse behaviour is the + // thing under test in approveSchemaDocColumn_collapsesLegacyDuplicates. + deduplicator = new SchemaDocumentationDeduplicator(schemaDocRepository, suggestionRepository, trainingService); applier = new CodeSuggestionApplier( suggestionRepository, schemaDocRepository, companyKnowledgeService, trainingService, - schemaScannerService + schemaScannerService, + deduplicator ); } @@ -57,7 +65,7 @@ void approveSchemaDocColumn_writesCodeDerivedRow() throws Exception { suggestion.setTargetObject("fct_ashram_visit.party_size"); suggestion.setPayload(Map.of("objectKind", "COLUMN", "businessTerms", List.of("party size"))); - when(suggestionRepository.findById("s1")).thenReturn(Optional.of(suggestion)); + when(suggestionRepository.findByIdForUpdate("s1")).thenReturn(Optional.of(suggestion)); when(suggestionRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); when(schemaDocRepository.findByConnectionIdAndObjectTypeAndObjectNameAndParentObjectAndSource( eq("conn-1"), @@ -65,7 +73,7 @@ void approveSchemaDocColumn_writesCodeDerivedRow() throws Exception { eq("party_size"), eq("acme_erp.fct_ashram_visit"), eq(DocumentationSource.CODE_DERIVED) - )).thenReturn(Optional.empty()); + )).thenReturn(List.of()); SchemaMetadata schema = new SchemaMetadata(); schema.setDatabaseName("acme_erp"); when(schemaScannerService.scanSchema("conn-1")).thenReturn(schema); @@ -90,12 +98,143 @@ void approveSchemaDocColumn_writesCodeDerivedRow() throws Exception { verify(trainingService).upsertDocumentationEmbedding(saved); } + /** + * Regression: two identical CODE_DERIVED rows (written by a bulk approve + * submitted twice concurrently before V116's unique index existed) made the + * Optional-returning finder throw "Query did not return a unique result", + * which bulk-decide swallowed into "Approved 0 of N" forever. + */ + @Test + void approveSchemaDocColumn_collapsesLegacyDuplicates() throws Exception { + CodeKnowledgeSuggestion suggestion = baseSuggestion(); + suggestion.setTargetKind(CodeKnowledgeSuggestion.TargetKind.SCHEMA_DOC); + suggestion.setTargetObject("fct_ashram_visit.party_size"); + suggestion.setPayload(Map.of("objectKind", "COLUMN")); + + SchemaDocumentation older = duplicateRow("doc-old", LocalDateTime.of(2026, 8, 13, 17, 11, 28)); + SchemaDocumentation newer = duplicateRow("doc-new", LocalDateTime.of(2026, 8, 13, 17, 11, 31)); + + when(suggestionRepository.findByIdForUpdate("s1")).thenReturn(Optional.of(suggestion)); + when(suggestionRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); + when(schemaDocRepository.findByConnectionIdAndObjectTypeAndObjectNameAndParentObjectAndSource( + eq("conn-1"), + eq(SchemaDocumentation.DocumentationType.COLUMN), + eq("party_size"), + eq("acme_erp.fct_ashram_visit"), + eq(DocumentationSource.CODE_DERIVED) + )).thenReturn(List.of(older, newer)); + SchemaMetadata schema = new SchemaMetadata(); + schema.setDatabaseName("acme_erp"); + when(schemaScannerService.scanSchema("conn-1")).thenReturn(schema); + when(schemaDocRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); + + CodeKnowledgeSuggestion approved = applier.approve("s1", "admin", null); + + assertThat(approved.getStatus()).isEqualTo(CodeKnowledgeSuggestion.Status.APPROVED); + // Newest survives — it is the row prior approvals linked via applied_doc_id. + assertThat(approved.getAppliedDocId()).isEqualTo("doc-new"); + verify(schemaDocRepository).delete(older); + verify(schemaDocRepository, never()).delete(newer); + verify(trainingService).deleteDocumentationEmbedding("conn-1", "doc-old"); + // Another approval pointing at the deleted row must follow the survivor. + verify(suggestionRepository).repointAppliedDocId("doc-new", List.of("doc-old")); + + ArgumentCaptor captor = ArgumentCaptor.forClass(SchemaDocumentation.class); + verify(schemaDocRepository).save(captor.capture()); + assertThat(captor.getValue().getId()).isEqualTo("doc-new"); + assertThat(captor.getValue().getDescription()) + .isEqualTo("Number of people in the visiting party."); + } + + /** A failed embedding delete must not abort the approval it is cleaning up for. */ + @Test + void approveSchemaDocColumn_collapseSurvivesEmbeddingDeleteFailure() throws Exception { + CodeKnowledgeSuggestion suggestion = baseSuggestion(); + suggestion.setTargetKind(CodeKnowledgeSuggestion.TargetKind.SCHEMA_DOC); + suggestion.setTargetObject("fct_ashram_visit.party_size"); + suggestion.setPayload(Map.of("objectKind", "COLUMN")); + + SchemaDocumentation older = duplicateRow("doc-old", LocalDateTime.of(2026, 8, 13, 17, 11, 28)); + SchemaDocumentation newer = duplicateRow("doc-new", LocalDateTime.of(2026, 8, 13, 17, 11, 31)); + + when(suggestionRepository.findByIdForUpdate("s1")).thenReturn(Optional.of(suggestion)); + when(suggestionRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); + when(schemaDocRepository.findByConnectionIdAndObjectTypeAndObjectNameAndParentObjectAndSource( + any(), any(), any(), any(), any())).thenReturn(List.of(older, newer)); + SchemaMetadata schema = new SchemaMetadata(); + schema.setDatabaseName("acme_erp"); + when(schemaScannerService.scanSchema("conn-1")).thenReturn(schema); + when(schemaDocRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); + doThrow(new RuntimeException("vector store down")) + .when(trainingService).deleteDocumentationEmbedding("conn-1", "doc-old"); + + CodeKnowledgeSuggestion approved = applier.approve("s1", "admin", null); + + assertThat(approved.getStatus()).isEqualTo(CodeKnowledgeSuggestion.Status.APPROVED); + verify(schemaDocRepository).delete(older); + } + + /** TABLE targets take the no-parent finder and must collapse the same way. */ + @Test + void approveSchemaDocTable_collapsesLegacyDuplicates() throws Exception { + CodeKnowledgeSuggestion suggestion = baseSuggestion(); + suggestion.setTargetKind(CodeKnowledgeSuggestion.TargetKind.SCHEMA_DOC); + suggestion.setTargetObject("fct_ashram_visit"); + suggestion.setPayload(Map.of("objectKind", "TABLE")); + + SchemaDocumentation older = duplicateRow("tbl-old", LocalDateTime.of(2026, 8, 13, 17, 11, 28)); + older.setObjectType(SchemaDocumentation.DocumentationType.TABLE); + older.setParentObject(null); + SchemaDocumentation newer = duplicateRow("tbl-new", LocalDateTime.of(2026, 8, 13, 17, 11, 31)); + newer.setObjectType(SchemaDocumentation.DocumentationType.TABLE); + newer.setParentObject(null); + + when(suggestionRepository.findByIdForUpdate("s1")).thenReturn(Optional.of(suggestion)); + when(suggestionRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); + when(schemaDocRepository.findByConnectionIdAndObjectTypeAndObjectNameAndSource( + eq("conn-1"), + eq(SchemaDocumentation.DocumentationType.TABLE), + eq("fct_ashram_visit"), + eq(DocumentationSource.CODE_DERIVED) + )).thenReturn(List.of(older, newer)); + when(schemaDocRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); + + CodeKnowledgeSuggestion approved = applier.approve("s1", "admin", null); + + assertThat(approved.getAppliedDocId()).isEqualTo("tbl-new"); + verify(schemaDocRepository).delete(older); + } + + /** Approve/reject must load under a row lock, else concurrent bulk decides re-create the duplicates. */ + @Test + void approveAndReject_loadTheSuggestionForUpdate() { + CodeKnowledgeSuggestion suggestion = baseSuggestion(); + suggestion.setStatus(CodeKnowledgeSuggestion.Status.APPROVED); + when(suggestionRepository.findByIdForUpdate("s1")).thenReturn(Optional.of(suggestion)); + + applier.approve("s1", "admin", null); + verify(suggestionRepository, never()).findById(any()); + } + + private static SchemaDocumentation duplicateRow(String id, LocalDateTime createdAt) { + SchemaDocumentation doc = new SchemaDocumentation(); + doc.setId(id); + doc.setConnectionId("conn-1"); + doc.setObjectType(SchemaDocumentation.DocumentationType.COLUMN); + doc.setObjectName("party_size"); + doc.setParentObject("acme_erp.fct_ashram_visit"); + doc.setSource(DocumentationSource.CODE_DERIVED); + doc.setDescription("stale"); + doc.setCreatedAt(createdAt); + return doc; + } + @Test void approveSchemaDoc_blankTarget_throwsInsteadOfSilentSkip() { CodeKnowledgeSuggestion suggestion = baseSuggestion(); suggestion.setTargetKind(CodeKnowledgeSuggestion.TargetKind.SCHEMA_DOC); suggestion.setTargetObject(" "); - when(suggestionRepository.findById("s1")).thenReturn(Optional.of(suggestion)); + when(suggestionRepository.findByIdForUpdate("s1")).thenReturn(Optional.of(suggestion)); assertThatThrownBy(() -> applier.approve("s1", "admin", null)) .isInstanceOf(IllegalArgumentException.class) @@ -108,7 +247,7 @@ void approveSchemaDoc_blankTarget_throwsInsteadOfSilentSkip() { void approveRejectedSuggestion_isRejected() { CodeKnowledgeSuggestion suggestion = baseSuggestion(); suggestion.setStatus(CodeKnowledgeSuggestion.Status.REJECTED); - when(suggestionRepository.findById("s1")).thenReturn(Optional.of(suggestion)); + when(suggestionRepository.findByIdForUpdate("s1")).thenReturn(Optional.of(suggestion)); assertThatThrownBy(() -> applier.approve("s1", "admin", null)) .isInstanceOf(IllegalStateException.class) diff --git a/scripts/self-host/e2e-review-approvals.py b/scripts/self-host/e2e-review-approvals.py index 1a792ed..23ed215 100755 --- a/scripts/self-host/e2e-review-approvals.py +++ b/scripts/self-host/e2e-review-approvals.py @@ -9,13 +9,16 @@ import json import os -import subprocess import sys import urllib.error from http.cookiejar import CookieJar from pathlib import Path +from concurrent.futures import ThreadPoolExecutor from urllib.request import HTTPCookieProcessor, Request, build_opener +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import vaultdb # noqa: E402 + ROOT = Path(__file__).resolve().parents[2] ENV = ROOT / ".env" @@ -93,14 +96,10 @@ def check(name: str, ok: bool, detail: str = ""): check("has KNOWLEDGE_ENTRY pending", len(knowledge) >= 1, f"n={len(knowledge)}") # ── 2. Constraint bootstrap: CODE_DERIVED must be allowed ────────────── - constraint = subprocess.check_output( - [ - "sudo", "-u", "postgres", "psql", "-d", "dba_agent", "-At", "-c", - "SELECT pg_get_constraintdef(oid) FROM pg_constraint " - "WHERE conname='schema_documentation_source_check'", - ], - text=True, - ).strip() + constraint = vaultdb.query( + "SELECT pg_get_constraintdef(oid) FROM pg_constraint " + "WHERE conname='schema_documentation_source_check'" + ) check("CODE_DERIVED in source CHECK", "CODE_DERIVED" in constraint, constraint) # ── 3. Approve SCHEMA_DOC (the customer failure path) ────────────────── @@ -195,22 +194,25 @@ def check(name: str, ok: bool, detail: str = ""): check("knowledge approvals applied", len(knowledge_ok) >= 1, f"n={len(knowledge_ok)}") # ── 10. Simulate missing CODE_DERIVED then ensure repair path works ──── - # Existing CODE_DERIVED rows block shrinking the CHECK, so park them first. - subprocess.check_call([ - "sudo", "-u", "postgres", "psql", "-d", "dba_agent", "-c", - "UPDATE schema_documentation SET source='USER' WHERE source='CODE_DERIVED'; " + # Existing CODE_DERIVED rows block shrinking the CHECK, so park them first — + # in a scratch table, not by rewriting source in place. The original version + # flipped every real CODE_DERIVED row to USER and never restored it, so a run + # against a live install silently relabelled the user's approved code-derived + # docs; it would also collide with V116's unique index the moment a user-written + # doc existed for the same column. + vaultdb.execute( + "DROP TABLE IF EXISTS e2e_parked_code_derived; " + "CREATE TABLE e2e_parked_code_derived AS " + " SELECT * FROM schema_documentation WHERE source='CODE_DERIVED'; " + "DELETE FROM schema_documentation WHERE source='CODE_DERIVED'; " "ALTER TABLE schema_documentation DROP CONSTRAINT IF EXISTS schema_documentation_source_check; " "ALTER TABLE schema_documentation ADD CONSTRAINT schema_documentation_source_check " "CHECK (source::text = ANY (ARRAY['USER','AI_GENERATED','CSV_IMPORT']::text[]));", - ]) - broken = subprocess.check_output( - [ - "sudo", "-u", "postgres", "psql", "-d", "dba_agent", "-At", "-c", - "SELECT pg_get_constraintdef(oid) FROM pg_constraint " - "WHERE conname='schema_documentation_source_check'", - ], - text=True, - ).strip() + ) + broken = vaultdb.query( + "SELECT pg_get_constraintdef(oid) FROM pg_constraint " + "WHERE conname='schema_documentation_source_check'" + ) check("can break CHECK for simulation", "CODE_DERIVED" not in broken) page6 = req(f"{backend}/code-scan/suggestions?connectionId={conn}&status=PENDING&page=0&size=20") @@ -233,21 +235,37 @@ def check(name: str, ok: bool, detail: str = ""): else: check("broken CHECK yields failed bulk with error detail", False, "no pending SCHEMA_DOC") - # Repair CHECK the same way the startup initializer does (without restart). - subprocess.check_call([ - "sudo", "-u", "postgres", "psql", "-d", "dba_agent", "-c", + # Repair CHECK the same way the startup initializer does (without restart), + # then un-park every row the simulation removed. The NOT EXISTS guard skips a + # row the post-repair retry has meanwhile recreated under the same key, which + # the unique index would otherwise reject. + vaultdb.execute( "ALTER TABLE schema_documentation DROP CONSTRAINT IF EXISTS schema_documentation_source_check; " "ALTER TABLE schema_documentation ADD CONSTRAINT schema_documentation_source_check " "CHECK (source::text = ANY (ARRAY['USER','AI_GENERATED','CSV_IMPORT','CODE_DERIVED']::text[]));", - ]) - repaired = subprocess.check_output( - [ - "sudo", "-u", "postgres", "psql", "-d", "dba_agent", "-At", "-c", - "SELECT pg_get_constraintdef(oid) FROM pg_constraint " - "WHERE conname='schema_documentation_source_check'", - ], - text=True, - ).strip() + ) + parked_count = vaultdb.query("SELECT count(*) FROM e2e_parked_code_derived") + vaultdb.execute( + "INSERT INTO schema_documentation SELECT p.* FROM e2e_parked_code_derived p " + "WHERE NOT EXISTS (SELECT 1 FROM schema_documentation d WHERE d.id = p.id) " + " AND NOT EXISTS (SELECT 1 FROM schema_documentation d " + " WHERE d.connection_id = p.connection_id " + " AND d.object_type = p.object_type " + " AND d.object_name = p.object_name " + " AND coalesce(d.parent_object,'') = coalesce(p.parent_object,'') " + " AND d.source = p.source); " + "DROP TABLE e2e_parked_code_derived;", + ) + restored = vaultdb.query("SELECT count(*) FROM schema_documentation WHERE source='CODE_DERIVED'") + check( + "parked CODE_DERIVED rows restored", + int(restored) >= int(parked_count), + f"parked={parked_count} now={restored}", + ) + repaired = vaultdb.query( + "SELECT pg_get_constraintdef(oid) FROM pg_constraint " + "WHERE conname='schema_documentation_source_check'" + ) check("CHECK repaired with CODE_DERIVED", "CODE_DERIVED" in repaired) if schema_pending: @@ -261,6 +279,169 @@ def check(name: str, ok: bool, detail: str = ""): f"status={retry.get('status')}", ) + # ── 11. Duplicate schema_documentation rows must not wedge approve ───── + # The customer's "APPROVED 0 OF 2" was two IncorrectResultSizeDataAccessException + # ("Query did not return a unique result: 2 results were returned") thrown by the + # upsert lookup in CodeSuggestionApplier.applySchemaDoc against duplicate rows a + # double-submitted bulk approve had left behind. + # + # Approve once to learn the exact key the applier writes (it database-qualifies + # parent_object via resolveDatabaseName, so guessing it from existing rows picks + # the wrong prefix and the test silently exercises nothing), then reset the + # suggestion, plant an older duplicate under that key, and approve again. + page7 = req(f"{backend}/code-scan/suggestions?connectionId={conn}&status=PENDING&page=0&size=50") + dup_target = next( + (s for s in (page7.get("content") or []) + if s.get("targetKind") == "SCHEMA_DOC" and "." in (s.get("targetObject") or "")), + None, + ) + if not dup_target: + check("duplicate rows collapse on approve", False, "no pending column SCHEMA_DOC") + else: + first = req( + f"{backend}/code-scan/suggestions/{dup_target['id']}/decide?connectionId={conn}", + {"decision": "APPROVED"}, + ) + doc_id = first.get("appliedDocId") + check("duplicate-test baseline approve", bool(doc_id), json.dumps(first)[:120]) + + key = vaultdb.query( + "SELECT object_name || '|' || coalesce(parent_object,'') " + f"FROM schema_documentation WHERE id='{doc_id}'" + ) + column, _, qualified = key.partition("|") + # Anchor the planted row's timestamp to the real one: approve UPDATES the + # row an earlier scan wrote, so its created_at is historical, and a + # hardcoded "old" date can easily be the newer of the two. + baseline_created = vaultdb.query( + f"SELECT created_at FROM schema_documentation WHERE id='{doc_id}'" + ) + where = ( + f"connection_id='{conn}' AND object_type='COLUMN' AND object_name='{column}' " + f"AND coalesce(parent_object,'')='{qualified}' AND source='CODE_DERIVED'" + ) + + # The unique index is what stops this happening for real, so drop it for the + # simulation. restore_unique_index() runs even if an assertion below throws — + # leaving the install without the constraint would be worse than a failed test. + vaultdb.execute("DROP INDEX IF EXISTS ux_schema_doc_target") + try: + vaultdb.execute( + "INSERT INTO schema_documentation " + "(id, connection_id, object_type, object_name, parent_object, description, " + " source, created_at) " + f"VALUES ('e2e-dup-older', '{conn}', 'COLUMN', '{column}', '{qualified}', " + f"'stale duplicate', 'CODE_DERIVED', TIMESTAMP '{baseline_created}' - INTERVAL '1 hour')" + ) + vaultdb.execute( + "UPDATE code_knowledge_suggestion SET status='PENDING', applied_doc_id=NULL, " + f"decided_at=NULL, decided_by=NULL WHERE id='{dup_target['id']}'" + ) + # A second, already-approved suggestion pointing at the row that is about + # to be deleted — its reference must follow the survivor, not dangle. + vaultdb.execute( + "UPDATE code_knowledge_suggestion SET applied_doc_id='e2e-dup-older' " + f"WHERE id = (SELECT id FROM code_knowledge_suggestion WHERE connection_id='{conn}' " + f" AND status='APPROVED' AND id <> '{dup_target['id']}' LIMIT 1)" + ) + before = vaultdb.query(f"SELECT count(*) FROM schema_documentation WHERE {where}") + check("duplicate pair seeded", before == "2", f"rows={before}") + + dup_bulk = req( + f"{backend}/code-scan/suggestions/bulk-decide?connectionId={conn}", + {"ids": [dup_target["id"]], "decision": "APPROVED"}, + ) + check( + "duplicate rows collapse on approve", + dup_bulk.get("succeeded") == 1 and dup_bulk.get("failed") == 0, + json.dumps(dup_bulk), + ) + after = vaultdb.query(f"SELECT count(*) FROM schema_documentation WHERE {where}") + check("collapsed to a single row", after == "1", f"rows={after}") + survivor = vaultdb.query(f"SELECT id FROM schema_documentation WHERE {where}") + # Newest wins — the row applied_doc_id already pointed at. + check("newest duplicate survived", survivor == doc_id, f"{survivor} vs {doc_id}") + content = vaultdb.query( + f"SELECT description FROM schema_documentation WHERE id='{survivor}'" + ) + check("survivor holds approved content", "stale duplicate" not in content, content[:60]) + gone = vaultdb.query( + "SELECT count(*) FROM schema_documentation WHERE id='e2e-dup-older'" + ) + check("older duplicate deleted", gone == "0", gone) + embedding = vaultdb.query("SELECT count(*) FROM rag_documents WHERE id='e2e-dup-older'") + check("loser embedding removed", embedding == "0", embedding) + dangling = vaultdb.query( + "SELECT count(*) FROM code_knowledge_suggestion WHERE applied_doc_id='e2e-dup-older'" + ) + check("applied_doc_id repointed off the deleted row", dangling == "0", dangling) + orphans = vaultdb.query( + "SELECT count(*) FROM code_knowledge_suggestion s WHERE s.applied_doc_id IS NOT NULL " + "AND NOT EXISTS (SELECT 1 FROM schema_documentation d WHERE d.id = s.applied_doc_id)" + ) + check("no dangling applied_doc_id anywhere", orphans == "0", orphans) + finally: + # Only ever remove the planted row, and only while nothing points at it. + # An earlier version deleted it unconditionally, which destroyed real + # approved content on the run where collapse had chosen it as survivor. + vaultdb.execute( + "DELETE FROM schema_documentation d WHERE d.id='e2e-dup-older' " + "AND NOT EXISTS (SELECT 1 FROM code_knowledge_suggestion s " + " WHERE s.applied_doc_id = d.id)" + ) + vaultdb.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS ux_schema_doc_target ON schema_documentation " + "(connection_id, object_type, object_name, coalesce(parent_object, ''), source)" + ) + + # ── 12. The unique index refuses a second row for the same key ───────── + idx = vaultdb.query( + "SELECT count(*) FROM pg_indexes WHERE indexname='ux_schema_doc_target'" + ) + check("unique index present", idx == "1", idx) + try: + vaultdb.execute( + "INSERT INTO schema_documentation " + "(id, connection_id, object_type, object_name, parent_object, description, source, created_at) " + "SELECT 'e2e-dup-clash', connection_id, object_type, object_name, parent_object, " + " description, source, now() " + "FROM schema_documentation LIMIT 1" + ) + vaultdb.execute("DELETE FROM schema_documentation WHERE id='e2e-dup-clash'") + check("unique index rejects a duplicate key", False, "insert succeeded") + except Exception as e: # noqa: BLE001 - psql exiting non-zero IS the assertion + check("unique index rejects a duplicate key", True, type(e).__name__) + + # ── 13. Concurrent approve of one suggestion writes exactly one doc row ─ + # Two bulk approves racing on the same PENDING row is what created the + # duplicates in the first place; approve now loads the suggestion FOR UPDATE. + page8 = req(f"{backend}/code-scan/suggestions?connectionId={conn}&status=PENDING&page=0&size=50") + race_target = next( + (s for s in (page8.get("content") or []) + if s.get("targetKind") == "SCHEMA_DOC" and "." in (s.get("targetObject") or "")), + None, + ) + if not race_target: + check("concurrent approve writes one row", False, "no pending column SCHEMA_DOC") + else: + url = f"{backend}/code-scan/suggestions/bulk-decide?connectionId={conn}" + payload = {"ids": [race_target["id"]], "decision": "APPROVED"} + with ThreadPoolExecutor(max_workers=2) as pool: + results = [f.result() for f in + [pool.submit(req, url, payload), pool.submit(req, url, payload)]] + parent, _, column = race_target["targetObject"].partition(".") + rows = vaultdb.query( + f"SELECT count(*) FROM schema_documentation WHERE connection_id='{conn}' " + f"AND object_type='COLUMN' AND object_name='{column}' " + f"AND parent_object LIKE '%{parent}' AND source='CODE_DERIVED'" + ) + check("concurrent approve writes one row", rows == "1", f"rows={rows}") + check( + "neither concurrent approve reported an error", + all(r.get("failed") == 0 for r in results), + json.dumps(results), + ) + print() if failures: print(f"✗ {len(failures)} failure(s): {failures}", file=sys.stderr) diff --git a/scripts/self-host/seed-review-suggestions.py b/scripts/self-host/seed-review-suggestions.py index dd302c5..f062190 100755 --- a/scripts/self-host/seed-review-suggestions.py +++ b/scripts/self-host/seed-review-suggestions.py @@ -12,6 +12,8 @@ import argparse import json import os +import shutil +import subprocess import sys import uuid from datetime import datetime, timezone @@ -37,6 +39,45 @@ def load_env(path: Path) -> dict[str, str]: return out +def apply_sql(sql_path: Path) -> int: + """Run the seed SQL against the vault DB, host install or Compose. + + A Compose deployment has no host-side postgres role (or psql at all), so the + original `sudo -u postgres psql` path only ever worked on a bare-metal + install — the verify command in the PR that added this script failed on the + topology `scripts/self-host/install.sh` actually produces. + """ + redirect = ">/tmp/seed-review.out 2>&1" + if shutil.which("psql"): + rc = os.system( + f"sudo -u postgres psql -d dba_agent -v ON_ERROR_STOP=1 -f {sql_path} {redirect}" + ) + if rc == 0: + return 0 + print("→ host psql failed; trying the Compose postgres container", file=sys.stderr) + + container = compose_postgres_container() + if not container: + print("No psql on PATH and no running Compose postgres container found.", file=sys.stderr) + return 1 + print(f"→ applying via docker exec {container}") + return os.system( + f"docker exec -i {container} psql -U postgres -d dba_agent -v ON_ERROR_STOP=1 " + f"< {sql_path} {redirect}" + ) + + +def compose_postgres_container() -> str | None: + """Name of the running postgres container, whatever the compose project is.""" + out = subprocess.run( + ["docker", "ps", "--filter", "label=com.docker.compose.service=postgres", + "--format", "{{.Names}}"], + capture_output=True, text=True, check=False, + ) + names = [n for n in out.stdout.split() if n] + return names[0] if names else None + + def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("connection_id", nargs="?") @@ -155,7 +196,7 @@ def req(url: str, data=None, method: str | None = None): sql_path = Path("/tmp/seed-review-suggestions.sql") sql_path.write_text(sql) print(f"→ writing {sql_path} ({args.count} suggestions)") - rc = os.system(f"sudo -u postgres psql -d dba_agent -v ON_ERROR_STOP=1 -f {sql_path} >/tmp/seed-review.out 2>&1") + rc = apply_sql(sql_path) if rc != 0: print(open("/tmp/seed-review.out").read(), file=sys.stderr) return 1 diff --git a/scripts/self-host/vaultdb.py b/scripts/self-host/vaultdb.py new file mode 100644 index 0000000..8201374 --- /dev/null +++ b/scripts/self-host/vaultdb.py @@ -0,0 +1,55 @@ +"""Run SQL against the DeepSQL vault DB from a self-host script. + +`sudo -u postgres psql` only works on a bare-metal install. `install.sh` produces +a Compose deployment where Postgres is a container and the host has no postgres +role — often no psql at all — so scripts that hardcode that command fail on the +topology they are meant to verify. Resolve the path once, here. +""" +from __future__ import annotations + +import shutil +import subprocess + + +def _host_psql(args: list[str]) -> list[str]: + return ["sudo", "-u", "postgres", "psql", "-d", "dba_agent", *args] + + +def _container_psql(container: str, args: list[str]) -> list[str]: + return ["docker", "exec", "-i", container, "psql", "-U", "postgres", "-d", "dba_agent", *args] + + +def postgres_container() -> str | None: + """Name of the running Compose postgres container, whatever the project is.""" + out = subprocess.run( + ["docker", "ps", "--filter", "label=com.docker.compose.service=postgres", + "--format", "{{.Names}}"], + capture_output=True, text=True, check=False, + ) + names = [n for n in out.stdout.split() if n] + return names[0] if names else None + + +def _command(args: list[str]) -> list[str]: + if shutil.which("psql"): + probe = subprocess.run(_host_psql(["-At", "-c", "SELECT 1"]), + capture_output=True, text=True, check=False) + if probe.returncode == 0: + return _host_psql(args) + container = postgres_container() + if not container: + raise RuntimeError( + "Cannot reach the vault DB: no usable host psql and no running " + "Compose postgres container." + ) + return _container_psql(container, args) + + +def query(sql: str) -> str: + """Run SQL and return stdout, raising on failure.""" + return subprocess.check_output(_command(["-At", "-c", sql]), text=True).strip() + + +def execute(sql: str) -> None: + """Run SQL for effect, raising on failure.""" + subprocess.check_call(_command(["-v", "ON_ERROR_STOP=1", "-c", sql])) diff --git a/src/lib/hooks/queries/useCodeScan.js b/src/lib/hooks/queries/useCodeScan.js index 18ffbf2..10e4ed8 100644 --- a/src/lib/hooks/queries/useCodeScan.js +++ b/src/lib/hooks/queries/useCodeScan.js @@ -149,15 +149,33 @@ export function useAllCodeScanSuggestions({ connectionId, status = 'PENDING' }) }) } +/** + * Everything an approve/reject writes, in one place. + * + * A decision is not confined to the suggestion row: approving a SCHEMA_DOC + * upserts `schema_documentation` (served by `brain/notes`, which backs the Write + * notes tab and its coverage counts) and re-embeds it, and approving a + * KNOWLEDGE_ENTRY creates a company knowledge entry. Invalidating only codeScan + + * companyKnowledge left every schema-doc-derived count stale until a page reload. + */ +function invalidateAfterDecision(queryClient, connectionId) { + if (!connectionId) return + ;[ + queryKeys.codeScan.all(connectionId), + queryKeys.companyKnowledge.all(connectionId), + // schema_documentation — brain/notes, coverage counts, understanding. + queryKeys.brain.all(connectionId), + // Accepting a description can resolve an ambiguity flagged in Unresolved. + queryKeys.schemaContext.all(connectionId), + ].forEach((queryKey) => queryClient.invalidateQueries({ queryKey })) +} + export function useDecideCodeScanSuggestion() { const queryClient = useQueryClient() return useMutation({ mutationFn: codeScanAPI.decide, onSuccess: (_data, variables) => { - if (variables?.connectionId) { - queryClient.invalidateQueries({ queryKey: queryKeys.codeScan.all(variables.connectionId) }) - queryClient.invalidateQueries({ queryKey: queryKeys.companyKnowledge.all(variables.connectionId) }) - } + invalidateAfterDecision(queryClient, variables?.connectionId) }, }) } @@ -167,10 +185,7 @@ export function useBulkDecideCodeScanSuggestions() { return useMutation({ mutationFn: codeScanAPI.bulkDecide, onSuccess: (_data, variables) => { - if (variables?.connectionId) { - queryClient.invalidateQueries({ queryKey: queryKeys.codeScan.all(variables.connectionId) }) - queryClient.invalidateQueries({ queryKey: queryKeys.companyKnowledge.all(variables.connectionId) }) - } + invalidateAfterDecision(queryClient, variables?.connectionId) }, }) }