feat!(backend): refactor multi-segment submission (2/n) (#5398)

resolves #4708,
#4734

partially resolves
#5392,
#5185 (comment)

Builds on #5382

### BREAKING CHANGES

When users submit to multi-segmented organisms and want to group
multiple segments under one metadata entry they are now required to add
an additional `fastaId` column with a space (or comma) -separated list
of the `fastaIds` (fasta header IDs) of the respective sequences. If no
`fastaId` column is supplied the `submissionId` will be used instead and
the backend will assume that (as in the single-segmented case) there is
a one-to-one mapping of metadata `submissionId` to `fastaId`.

This new submission structure was voted for in microbioinfo:
https://microbial-bioinfo.slack.com/archives/CB0HYT53M/p1760961465729399
and discussed in
https://app.nuclino.com/Loculus/Development/2025-10-20-Weekly-6d5fe89f-8ded-4286-b892-d215e0a498f6
(and in other meetings)

Nextclade sort will be used to assign segments/subtypes for all aligned
sequences:
```
minimizer_index: <url_to_minimizer_index_used_by_nextclade_sort>
```
For organisms without a nextclade dataset we still allow the fasta
headers to be used to determine the segment/subtype - entries must have
the format <submissionId>_<segmentName> (as in current set up).

As preprocessing now assigns segments it will return a map from the
segment (or subtype) to the fastaHeader in the processedData:
`sequenceNameToFastaHeaderMap`. This allows us to surface this
assignment on the edit page.

### Testing

You can use pathoplexus/dev_example_data#2 for
testing.

### Prepro config changes

Instead of having a dictionary for the nextclade datasets and servers we
make `nucleotideSequences` a list of sequences:
```
nextclade_dataset_name: 
    L: nextstrain/cchfv/linked/L
    M: nextstrain/cchfv/linked/M
    S: nextstrain/cchfv/linked/S
nextclade_dataset_server: https://raw.githubusercontent.com/nextstrain/nextclade_data/cornelius-cchfv/data_output
genes: [RdRp, GPC, NP]
```

```
nucleotideSequences:
  - name: L
    nextclade_dataset_name: nextstrain/cchfv/linked/L
    nextclade_dataset_tag: <optional - was previously incorrectly placed on an organism level> 
    nextclade_dataset_server: <optional overwrites nextclade_dataset_server for this seq>
    accepted_sort_matches: <optional, used for classify_with_nextclade_sort and require_nextclade_sort_match, if not given nextclade_dataset_name is used> 
    gene_prefix: <optional, prefix to add to genes produced by nextclade run, e.g. nextclade labels genes as `AV1` but we expect `EV1_AV1`, here `EV1` would be the prefix >
  - name: M
    nextclade_dataset_name: nextstrain/cchfv/linked/M
  - name: S
    nextclade_dataset_name: nextstrain/cchfv/linked/S
nextclade_dataset_server: https://raw.githubusercontent.com/nextstrain/nextclade_data/cornelius-cchfv/data_output
```

Note the templates now also generate the genes list from the merged
config.

### PR Checklist
- [ ] Update values.schema.json
- [x] keep tests for alignment NONE case
- [x] Create a minimizer for tests using:
https://github.com/loculus-project/nextclade-sort-minimizer-creator
- [x] Any manual testing that has been done is documented: submission of
EVs from test folder were submitted with the same fastaHeader as the
submissionId -> this succeeded, additionally the submission of CCHF with
a fastaID column in the metadata was tested (also in folder above),
additionally revision of a segment was tested
- [x] Have preprocessing send back a segment: fastaHeader mapping

## Future Work
- [ ] add integration testing for full EV submission user journey
- [ ] improve CCHF minimizer (some segments are again not assigned)
- [ ] discuss if the originalData dictionary should be migrated
(persistent DB has segmentName as key, now we have fastaHeader as key)
- [ ] update PPX docs with new multi-segment submission format

🚀 Preview: https://multi-segment-submission.loculus.org

---------

Co-authored-by: Cornelius Roemer <[email protected]>

Author: anna-parker

backend/docs/db/schema.sql

@@ -380,7 +380,8 @@ CREATE TABLE public.metadata_upload_aux_table (
     group_id integer,
     uploaded_at timestamp without time zone NOT NULL,
     metadata jsonb NOT NULL,
-    files jsonb
+    files jsonb,
+    fasta_ids jsonb DEFAULT '[]'::jsonb
 );
 
 
@@ -540,9 +541,8 @@ ALTER VIEW public.sequence_entries_view OWNER TO postgres;
 
 CREATE TABLE public.sequence_upload_aux_table (
     upload_id text NOT NULL,
-    submission_id text NOT NULL,
-    segment_name text NOT NULL,
-    compressed_sequence_data text NOT NULL
+    compressed_sequence_data text NOT NULL,
+    fasta_id text NOT NULL
 );
 
 
@@ -755,7 +755,7 @@ ALTER TABLE ONLY public.sequence_entries_preprocessed_data
 --
 
 ALTER TABLE ONLY public.sequence_upload_aux_table
-    ADD CONSTRAINT sequence_upload_aux_table_pkey PRIMARY KEY (upload_id, submission_id, segment_name);
+    ADD CONSTRAINT sequence_upload_aux_table_pkey PRIMARY KEY (upload_id, fasta_id);
 
 
 --
@@ -796,6 +796,13 @@ CREATE INDEX data_use_terms_table_accession_idx ON public.data_use_terms_table U
 CREATE INDEX flyway_schema_history_s_idx ON public.flyway_schema_history USING btree (success);
 
 
+--
+-- Name: metadata_upload_aux_table_fasta_ids_idx; Type: INDEX; Schema: public; Owner: postgres
+--
+
+CREATE INDEX metadata_upload_aux_table_fasta_ids_idx ON public.metadata_upload_aux_table USING gin (fasta_ids jsonb_path_ops);
+
+
 --
 -- Name: sequence_entries_organism_idx; Type: INDEX; Schema: public; Owner: postgres
 --

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.model.FastaId
 import org.loculus.backend.model.SubmissionId
 import org.loculus.backend.service.files.FileId
 import org.loculus.backend.utils.Accession
@@ -166,6 +167,11 @@ data class ProcessedData<SequenceType>(
         description = "The key is the gene name, the value is a list of amino acid insertions",
     )
     val aminoAcidInsertions: Map<GeneName, List<Insertion>>,
+    @Schema(
+        example = """{"segment1": "fastaHeader1", "segment2": "fastaHeader2"}""",
+        description = "The key is the segment name, the value is the fastaHeader of the original Data",
+    )
+    val sequenceNameToFastaHeaderMap: Map<SegmentName, String> = emptyMap(),
     @Schema(
         example = """{"raw_reads": [{"fileId": "s0m3-uUiDd", "name": "data.fastaq"}], "sequencing_logs": []}""",
         description = "The key is the file category name, the value is a list of files, with ID and name.",
@@ -300,9 +306,9 @@ data class OriginalDataInternal<SequenceType, FilesType>(
     val metadata: Map<String, String>,
     @Schema(
         example = "{\"segment1\": \"ACTG\", \"segment2\": \"GTCA\"}",
-        description = "The key is the segment name, the value is the nucleotide sequence",
+        description = "The key is the fastaID, the value is the nucleotide sequence",
     )
-    val unalignedNucleotideSequences: Map<SegmentName, SequenceType?>,
+    val unalignedNucleotideSequences: Map<FastaId, SequenceType?>,
     @Schema(
         example = """{"raw_reads": [{"fileId": "f1le-uuId-asdf", "name": "myfile.fastaq"]}""",
         description = "A map from file categories, to lists of files. The files can also have URLs.",

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

@@ -1,13 +1,13 @@
 package org.loculus.backend.controller
 
-import org.loculus.backend.model.HEADER_TO_CONNECT_METADATA_AND_SEQUENCES
+import org.loculus.backend.model.METADATA_ID_HEADER
 
 const val SUBMIT_RESPONSE_DESCRIPTION = """
 Returns a list of accession, version and submissionId of the submitted sequence entries. 
-The submissionId is the (locally unique) '$HEADER_TO_CONNECT_METADATA_AND_SEQUENCES' provided by the submitter in the metadata file. 
+The submissionId is the (locally unique) '$METADATA_ID_HEADER' provided by the submitter in the metadata file. 
 The version will be 1 for every sequence. 
 The accession is the (globally unique) id that the system assigned to the sequence entry. 
-You can use this response to associate the user provided $HEADER_TO_CONNECT_METADATA_AND_SEQUENCES with the system assigned accession.
+You can use this response to associate the user provided $METADATA_ID_HEADER with the system assigned accession.
 """
 
 const val SUBMIT_ERROR_RESPONSE = """
@@ -18,16 +18,18 @@ const val METADATA_FILE_DESCRIPTION = """
 A TSV (tab separated values) file containing the metadata of the submitted sequence entries. 
 The file may be compressed with zstd, xz, zip, gzip, lzma, bzip2 (with common extensions).
 It must contain the column names.
-The field '$HEADER_TO_CONNECT_METADATA_AND_SEQUENCES' is required and must be unique within the provided dataset.
+The field '$METADATA_ID_HEADER' is required and must be unique within the provided dataset.
 It is used to associate metadata to the sequences in the sequences fasta file.
 """
+
+// TODO: update description
 const val SEQUENCE_FILE_DESCRIPTION = """
 A fasta file containing the unaligned nucleotide sequences of the submitted sequences.
 The file may be compressed with zstd, xz, zip, gzip, lzma, bzip2 (with common extensions).
 If the underlying organism has a single segment,
-the headers of the fasta file must match the '$HEADER_TO_CONNECT_METADATA_AND_SEQUENCES' field in the metadata file.
+the headers of the fasta file must match the '$METADATA_ID_HEADER' field in the metadata file.
 If the underlying organism has multiple segments,
-the headers of the fasta file must be of the form '>[$HEADER_TO_CONNECT_METADATA_AND_SEQUENCES]_[segmentName]'.
+the headers of the fasta file must be of the form '>[$METADATA_ID_HEADER]_[segmentName]'.
 """
 
 const val FILE_MAPPING_DESCRIPTION = """
@@ -114,7 +116,7 @@ The version will increase by one in respect to the original accession version.
 
 const val REVISED_METADATA_FILE_DESCRIPTION = """
 A TSV (tab separated values) file containing the metadata of the revised data.
-The first row must contain the column names. The column '$HEADER_TO_CONNECT_METADATA_AND_SEQUENCES' is required and must be unique within the 
+The first row must contain the column names. The column '$METADATA_ID_HEADER' is required and must be unique within the 
 provided dataset. It is used to associate metadata to the sequences in the sequences fasta file.
 Additionally, the column 'accession' is required and must match the accession of the original sequence entry.
 """

backend/src/main/kotlin/org/loculus/backend/model/SubmitModel.kt

@@ -31,13 +31,15 @@ import java.io.BufferedInputStream
 import java.io.File
 import java.io.InputStream
 
-const val HEADER_TO_CONNECT_METADATA_AND_SEQUENCES = "id"
-const val HEADER_TO_CONNECT_METADATA_AND_SEQUENCES_ALTERNATE_FOR_BACKCOMPAT = "submissionId"
+const val METADATA_ID_HEADER = "id"
+const val METADATA_ID_HEADER_ALTERNATE_FOR_BACKCOMPAT = "submissionId"
+const val FASTA_ID_HEADER = "fastaId"
 
 const val ACCESSION_HEADER = "accession"
 private val log = KotlinLogging.logger { }
 
 typealias SubmissionId = String
+typealias FastaId = String
 typealias SegmentName = String
 
 const val UNIQUE_CONSTRAINT_VIOLATION_SQL_STATE = "23505"
@@ -128,8 +130,13 @@ class SubmitModel(
         val metadataSubmissionIds = uploadDatabaseService.getMetadataUploadSubmissionIds(uploadId).toSet()
         if (requiresConsensusSequenceFile(submissionParams.organism)) {
             log.debug { "Validating submission with uploadId $uploadId" }
-            val sequenceSubmissionIds = uploadDatabaseService.getSequenceUploadSubmissionIds(uploadId).toSet()
-            validateSubmissionIdSetsForConsensusSequences(metadataSubmissionIds, sequenceSubmissionIds)
+            val metadataFastaIds = uploadDatabaseService.getFastaIdsForMetadata(uploadId).flatten()
+            val metadataFastaIdsSet = metadataFastaIds.toSet()
+            if (metadataFastaIdsSet.size < metadataFastaIds.size) {
+                throw UnprocessableEntityException("Metadata file contains duplicate fastaIds.")
+            }
+            val sequenceFastaIds = uploadDatabaseService.getSequenceUploadSubmissionIds(uploadId).toSet()
+            validateSubmissionIdSetsForConsensusSequences(metadataFastaIdsSet, sequenceFastaIds)
         }
 
         if (submissionParams is SubmissionParams.RevisionSubmissionParams) {
@@ -177,38 +184,39 @@ class SubmitModel(
             metadataFileTypes,
             metadataTempFileToDelete,
         )
+        val addFastaId = requiresConsensusSequenceFile(submissionParams.organism)
         try {
-            uploadMetadata(uploadId, submissionParams, metadataStream, batchSize)
+            uploadMetadata(uploadId, submissionParams, metadataStream, batchSize, addFastaId = addFastaId)
         } finally {
             metadataTempFileToDelete.delete()
         }
 
         val sequenceFile = submissionParams.sequenceFile
         if (sequenceFile == null) {
-            if (requiresConsensusSequenceFile(submissionParams.organism)) {
+            if (addFastaId) {
                 throw BadRequestException(
                     "Submissions for organism ${submissionParams.organism.name} require a sequence file.",
                 )
             }
-        } else {
-            if (!requiresConsensusSequenceFile(submissionParams.organism)) {
-                throw BadRequestException(
-                    "Sequence uploads are not allowed for organism ${submissionParams.organism.name}.",
-                )
-            }
+            return
+        }
+        if (!addFastaId) {
+            throw BadRequestException(
+                "Sequence uploads are not allowed for organism ${submissionParams.organism.name}.",
+            )
+        }
 
-            val sequenceTempFileToDelete = MaybeFile()
-            try {
-                val sequenceStream = getStreamFromFile(
-                    sequenceFile,
-                    uploadId,
-                    sequenceFileTypes,
-                    sequenceTempFileToDelete,
-                )
-                uploadSequences(uploadId, sequenceStream, batchSize, submissionParams.organism)
-            } finally {
-                sequenceTempFileToDelete.delete()
-            }
+        val sequenceTempFileToDelete = MaybeFile()
+        try {
+            val sequenceStream = getStreamFromFile(
+                sequenceFile,
+                uploadId,
+                sequenceFileTypes,
+                sequenceTempFileToDelete,
+            )
+            uploadSequences(uploadId, sequenceStream, batchSize, submissionParams.organism)
+        } finally {
+            sequenceTempFileToDelete.delete()
         }
     }
 
@@ -254,6 +262,7 @@ class SubmitModel(
         submissionParams: SubmissionParams,
         metadataStream: InputStream,
         batchSize: Int,
+        addFastaId: Boolean,
     ) {
         log.debug {
             "intermediate storing uploaded metadata of type ${submissionParams.uploadType.name} " +
@@ -263,7 +272,7 @@ class SubmitModel(
         try {
             when (submissionParams) {
                 is SubmissionParams.OriginalSubmissionParams -> {
-                    metadataEntryStreamAsSequence(metadataStream)
+                    metadataEntryStreamAsSequence(metadataStream, addFastaId)
                         .chunked(batchSize)
                         .forEach { batch ->
                             uploadDatabaseService.batchInsertMetadataInAuxTable(
@@ -279,7 +288,7 @@ class SubmitModel(
                 }
 
                 is SubmissionParams.RevisionSubmissionParams -> {
-                    revisionEntryStreamAsSequence(metadataStream)
+                    revisionEntryStreamAsSequence(metadataStream, addFastaId)
                         .chunked(batchSize)
                         .forEach { batch ->
                             uploadDatabaseService.batchInsertRevisedMetadataInAuxTable(
@@ -354,14 +363,15 @@ class SubmitModel(
 
         if (metadataKeysNotInSequences.isNotEmpty() || sequenceKeysNotInMetadata.isNotEmpty()) {
             val metadataNotPresentErrorText = if (metadataKeysNotInSequences.isNotEmpty()) {
-                "Metadata file contains ${metadataKeysNotInSequences.size} ids that are not present " +
+                "Metadata file contains ${metadataKeysNotInSequences.size} FASTA ids that are not present " +
                     "in the sequence file: " + metadataKeysNotInSequences.toList().joinToString(limit = 10) + "; "
             } else {
                 ""
             }
             val sequenceNotPresentErrorText = if (sequenceKeysNotInMetadata.isNotEmpty()) {
-                "Sequence file contains ${sequenceKeysNotInMetadata.size} ids that are not present " +
-                    "in the metadata file: " + sequenceKeysNotInMetadata.toList().joinToString(limit = 10)
+                "Sequence file contains ${sequenceKeysNotInMetadata.size} FASTA ids that are not present " +
+                    "in the metadata file: " +
+                    sequenceKeysNotInMetadata.toList().joinToString(limit = 10)
             } else {
                 ""
             }

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

@@ -118,6 +118,7 @@ class CompressionService(
                 }
             },
             processedData.aminoAcidInsertions,
+            processedData.sequenceNameToFastaHeaderMap,
             processedData.files,
         )
 
@@ -144,6 +145,7 @@ class CompressionService(
             }
         },
         processedData.aminoAcidInsertions,
+        processedData.sequenceNameToFastaHeaderMap,
         processedData.files,
     )
 

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

@@ -20,6 +20,7 @@ class EmptyProcessedDataProvider(private val backendConfig: BackendConfig) {
             alignedAminoAcidSequences = referenceGenome.genes.map { it.name }.associateWith { null },
             nucleotideInsertions = referenceGenome.nucleotideSequences.map { it.name }.associateWith { emptyList() },
             aminoAcidInsertions = referenceGenome.genes.map { it.name }.associateWith { emptyList() },
+            sequenceNameToFastaHeaderMap = referenceGenome.nucleotideSequences.map { it.name }.associateWith { "" },
             files = null,
         )
     }

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

@@ -232,6 +232,11 @@ class ProcessedSequenceEntryValidator(private val schema: Schema, private val re
             "alignedNucleotideSequences",
         )
 
+        validateNoUnknownSegment(
+            processedData.sequenceNameToFastaHeaderMap,
+            "sequenceNameToFastaHeaderMap",
+        )
+
         validateNoUnknownSegment(
             processedData.unalignedNucleotideSequences,
             "unalignedNucleotideSequences",

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

@@ -461,6 +461,7 @@ class SubmissionDatabaseService(
         aminoAcidInsertions = processedData.aminoAcidInsertions.mapValues { (_, it) ->
             it.map { insertion -> insertion.copy(sequence = insertion.sequence.uppercase(Locale.US)) }
         },
+        sequenceNameToFastaHeaderMap = processedData.sequenceNameToFastaHeaderMap,
     )
 
     private fun validateExternalMetadata(
@@ -1233,7 +1234,7 @@ class SubmissionDatabaseService(
             .fetchSize(streamBatchSize)
             .asSequence()
             .map {
-                // Revoked sequences have no original metdadata, hence null can happen
+                // Revoked sequences have no original metadata, hence null can happen
                 @Suppress("USELESS_ELVIS")
                 val metadata = it[originalMetadata] ?: null
                 val selectedMetadata = fields?.associateWith { field -> metadata?.get(field) }