feat(ingest, backend): improve ingest curation warning - add diff (#6633)

resolves #3084

Adds the option to filter by AccessionVersion on the
`get-unprocessed-metadata` endpoint. Have ingest call this endpoint to
get all previous metadata (includes sequence hashes, so it will
indirectly tell us if a sequence has been changed) when we get into the
case where ingest would like to revise a curated sequence.

Ingest will then add a diff of the curated metadata and the metadata it
would like to revise to the notification so that curators can decide
better to to handle the situation.

### PR Checklist
- [x] All necessary documentation has been adapted for backend endpoints
- [x] The backend endpoint updates are covered by unit tests
- [x] Any manual testing that has been done is documented (i.e. what
exactly was tested?)
1. Ingest sequences from the INSDC with initial hash
2. Curate a sequence from the INSDC as a superuser (add a revision, do
not change hash)
3. Ingest ignores revision as the hash has not changed
4. Log into database and manually modify the hash of the curated entry
(unprocessed data not original data)
5. Run the ingest cronjob and verify that an informative message is sent
to the loculus notifications slack:
<img width="2254" height="204" alt="image"
src="https://github.com/user-attachments/assets/2e91f1b8-6b94-4801-9624-c6fc53cecaaf"
/>

🚀 Preview: https://curation-warning-add-diff.loculus.org

---------

Co-authored-by: Cornelius Roemer <cornelius.roemer@gmail.com>

Author: anna-parker

backend/src/main/kotlin/org/loculus/backend/api/SubmissionTypes.kt

@@ -8,6 +8,7 @@ import com.fasterxml.jackson.databind.JsonDeserializer
 import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
 import io.swagger.v3.oas.annotations.media.Schema
+import org.loculus.backend.controller.UnprocessableEntityException
 import org.loculus.backend.model.FastaId
 import org.loculus.backend.model.SubmissionId
 import org.loculus.backend.service.files.FileId
@@ -26,7 +27,32 @@ interface AccessionVersionInterface {
 }
 
 data class AccessionVersion(override val accession: Accession, override val version: Version) :
-    AccessionVersionInterface
+    AccessionVersionInterface {
+    companion object {
+        fun fromString(value: String): AccessionVersion {
+            val accession = value.substringBeforeLast('.', missingDelimiterValue = "")
+            val versionString = value.substringAfterLast('.', missingDelimiterValue = "")
+
+            if (accession.isEmpty() || versionString.isEmpty() || '.' in accession) {
+                throw UnprocessableEntityException(
+                    "Invalid accession version format '$value', expected 'accession.version'",
+                )
+            }
+
+            val version = versionString.toLongOrNull()
+                ?: throw UnprocessableEntityException(
+                    "Invalid version in accession version '$value', expected a number",
+                )
+            if (version < 1) {
+                throw UnprocessableEntityException(
+                    "Invalid version in accession version '$value', version must be at least 1",
+                )
+            }
+
+            return AccessionVersion(accession, version)
+        }
+    }
+}
 
 data class SubmissionIdMapping(
     override val accession: Accession,

backend/src/main/kotlin/org/loculus/backend/controller/SubmissionController.kt

@@ -433,6 +433,10 @@ open class SubmissionController(
         @Parameter(
             description = "The metadata fields that should be returned. If not provided, all fields are returned.",
         ) @RequestParam(required = false) fields: List<String>?,
+        @Parameter(
+            description = "Filter by accession versions in 'accession.version' format. " +
+                "If not provided, all accession versions are considered.",
+        ) @RequestParam(required = false) accessionVersionsFilter: List<String>?,
         @Parameter(
             description = "Filter by group ids. If not provided, all groups are considered.",
         ) @RequestParam(required = false) groupIdsFilter: List<Int>?,
@@ -448,11 +452,16 @@ open class SubmissionController(
             headers.add(HttpHeaders.CONTENT_ENCODING, compression.compressionName)
         }
 
+        val parsedAccessionVersions = accessionVersionsFilter?.takeIf { it.isNotEmpty() }?.map {
+            AccessionVersion.fromString(it)
+        }
+
         val totalRecords = submissionDatabaseService.countUnprocessedMetadata(
             authenticatedUser,
             organism,
             groupIdsFilter?.takeIf { it.isNotEmpty() },
             statusesFilter?.takeIf { it.isNotEmpty() },
+            parsedAccessionVersions,
         )
         headers.add(X_TOTAL_RECORDS, totalRecords.toString())
         // TODO(https://github.com/loculus-project/loculus/issues/2778)
@@ -468,6 +477,7 @@ open class SubmissionController(
                 groupIdsFilter?.takeIf { it.isNotEmpty() },
                 statusesFilter?.takeIf { it.isNotEmpty() },
                 fields?.takeIf { it.isNotEmpty() },
+                parsedAccessionVersions,
             )
         }
 

backend/src/main/kotlin/org/loculus/backend/service/seqsetcitations/SeqSetCitationsDatabaseService.kt

@@ -482,20 +482,7 @@ class SeqSetCitationsDatabaseService(
         }
         val accessionsWithVersions = mutableListOf<AccessionVersion>()
         for (record in seqSetRecords.filter { it.accession.contains('.') }) {
-            val parts = record.accession.split('.')
-            if (parts.size != 2) {
-                throw UnprocessableEntityException(
-                    "Invalid accession format '${record.accession}': expected format 'ACCESSION.VERSION'",
-                )
-            }
-            val versionString = parts[1]
-            val version = versionString.toLongOrNull()
-            if (version == null) {
-                throw UnprocessableEntityException(
-                    "Invalid version in accession '${record.accession}': '$versionString' is not a valid integer",
-                )
-            }
-            accessionsWithVersions.add(AccessionVersion(parts[0], version))
+            accessionsWithVersions.add(AccessionVersion.fromString(record.accession))
         }
 
         for (chunk in accessionsWithVersions.chunked(1000)) {

backend/src/main/kotlin/org/loculus/backend/service/submission/SubmissionDatabaseService.kt

@@ -1231,6 +1231,7 @@ class SubmissionDatabaseService(
         organism: Organism,
         groupIdsFilter: List<Int>?,
         statusesFilter: List<Status>?,
+        accessionVersionsFilter: List<AccessionVersion>?,
     ): Op<Boolean> {
         val organismCondition = SequenceEntriesView.organismIs(organism)
         val groupCondition = getGroupCondition(groupIdsFilter, authenticatedUser)
@@ -1239,7 +1240,12 @@ class SubmissionDatabaseService(
         } else {
             Op.TRUE
         }
-        val conditions = organismCondition and groupCondition and statusCondition
+        val accessionVersionCondition = if (accessionVersionsFilter != null) {
+            SequenceEntriesView.accessionVersionIsIn(accessionVersionsFilter)
+        } else {
+            Op.TRUE
+        }
+        val conditions = organismCondition and groupCondition and statusCondition and accessionVersionCondition
 
         return conditions
     }
@@ -1249,6 +1255,7 @@ class SubmissionDatabaseService(
         organism: Organism,
         groupIdsFilter: List<Int>?,
         statusesFilter: List<Status>?,
+        accessionVersionsFilter: List<AccessionVersion>?,
     ): Long = SequenceEntriesView
         .selectAll()
         .where(
@@ -1257,6 +1264,7 @@ class SubmissionDatabaseService(
                 organism,
                 groupIdsFilter,
                 statusesFilter,
+                accessionVersionsFilter,
             ),
         )
         .count()
@@ -1267,6 +1275,7 @@ class SubmissionDatabaseService(
         groupIdsFilter: List<Int>?,
         statusesFilter: List<Status>?,
         fields: List<String>?,
+        accessionVersionsFilter: List<AccessionVersion>?,
     ): Sequence<AccessionVersionUnprocessedMetadata> {
         val unprocessedMetadata = SequenceEntriesView.unprocessedDataColumn
             // It's actually <Map<String, String>?> but exposed does not support nullable types here
@@ -1287,6 +1296,7 @@ class SubmissionDatabaseService(
                     organism,
                     groupIdsFilter,
                     statusesFilter,
+                    accessionVersionsFilter,
                 ),
             )
             .fetchSize(streamBatchSize)

backend/src/test/kotlin/org/loculus/backend/api/SubmissionTypesTest.kt

@@ -0,0 +1,35 @@
+package org.loculus.backend.api
+
+import org.hamcrest.MatcherAssert.assertThat
+import org.hamcrest.Matchers.`is`
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import org.loculus.backend.controller.UnprocessableEntityException
+
+class SubmissionTypesTest {
+    @Test
+    fun `GIVEN valid accession version string THEN parses accession and version`() {
+        assertThat(AccessionVersion.fromString("LOC_123.42"), `is`(AccessionVersion("LOC_123", 42)))
+    }
+
+    @Test
+    fun `GIVEN accession version with invalid format THEN throws format exception`() {
+        listOf("LOC_123", ".1", "LOC_123.", "LOC.123.1").forEach {
+            assertThrows<UnprocessableEntityException> {
+                AccessionVersion.fromString(it)
+            }
+        }
+    }
+
+    @Test
+    fun `GIVEN non-numeric version THEN throws version exception`() {
+        val exception = assertThrows<UnprocessableEntityException> {
+            AccessionVersion.fromString("LOC_123.invalid")
+        }
+
+        assertThat(
+            exception.message,
+            `is`("Invalid version in accession version 'LOC_123.invalid', expected a number"),
+        )
+    }
+}

backend/src/test/kotlin/org/loculus/backend/controller/seqsetcitations/SeqSetValidationEndpointsTest.kt

@@ -67,7 +67,7 @@ class SeqSetValidationEndpointsTest(
                 jsonPath(
                     "\$.detail",
                     containsString(
-                        "Invalid version in accession 'ABCD.EF'",
+                        "Invalid version in accession version 'ABCD.EF', expected a number",
                     ),
                 ),
             )

backend/src/test/kotlin/org/loculus/backend/controller/submission/GetUnprocessedMetadataEndpointTest.kt

@@ -190,6 +190,29 @@ class GetUnprocessedMetadataEndpointTest(
             .andExpect(status().isOk)
     }
 
+    @Test
+    fun `WHEN I filter by accessionVersions THEN should return only those entries`() {
+        val entries = convenienceClient.prepareDefaultSequenceEntriesToApprovedForRelease()
+        val target = entries.first()
+        val accessionVersion = target.displayAccessionVersion()
+
+        val response = submissionControllerClient.getUnprocessedMetadata(
+            accessionVersionsFilter = listOf(accessionVersion),
+        )
+        response.andExpect(status().isOk)
+            .andExpect(header().string("x-total-records", `is`("1")))
+        val responseBody = response.expectNdjsonAndGetContent<AccessionVersionUnprocessedMetadata>()
+        assertThat(responseBody, hasSize(1))
+        assertThat(responseBody[0].displayAccessionVersion(), `is`(accessionVersion))
+    }
+
+    @Test
+    fun `WHEN I filter by accessionVersions with invalid format THEN returns 422`() {
+        submissionControllerClient.getUnprocessedMetadata(
+            accessionVersionsFilter = listOf("not-a-valid-accession-version"),
+        ).andExpect(status().isUnprocessableEntity)
+    }
+
     // Regression test for https://github.com/loculus-project/loculus/issues/4036
     @Test
     fun `GIVEN revoked sequences exist THEN endpoint does not throw exception`() {

backend/src/test/kotlin/org/loculus/backend/controller/submission/SubmissionControllerClient.kt

@@ -307,6 +307,7 @@ class SubmissionControllerClient(private val mockMvc: MockMvc, private val objec
         jwt: String? = jwtForDefaultUser,
         groupIdsFilter: List<Int>? = null,
         statusesFilter: List<Status>? = null,
+        accessionVersionsFilter: List<String>? = null,
         fields: List<String>? = null,
         compression: String? = null,
     ): ResultActions = mockMvc.perform(
@@ -320,6 +321,7 @@ class SubmissionControllerClient(private val mockMvc: MockMvc, private val objec
             }
             .param("groupIdsFilter", groupIdsFilter?.joinToString(",") { it.toString() })
             .param("statusesFilter", statusesFilter?.joinToString(",") { it.name })
+            .param("accessionVersionsFilter", accessionVersionsFilter?.joinToString(","))
             .param("fields", fields?.joinToString(",")),
     )
 

ingest/scripts/call_loculus.py

@@ -106,7 +106,15 @@ def record_factory(*args, **kwargs):
 
     if mode == "get-submitted":
         logger.info("Getting submitted sequences")
-        get_submitted(config, output)
+        if config.segmented:
+            insdc_key = [
+                "insdcAccessionBase" + "_" + segment for segment in config.nucleotide_sequences
+            ]
+        else:
+            insdc_key = ["insdcAccessionBase"]
+
+        fields = ["hash", *insdc_key]
+        get_submitted(config, output, fields)
 
 
 if __name__ == "__main__":