package dev.fitko.fitconnect.integrationtests;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.nimbusds.jose.jwk.JWK;
import com.nimbusds.jose.jwk.RSAKey;
import dev.fitko.fitconnect.api.config.ApplicationConfig;
import dev.fitko.fitconnect.api.config.chunking.AttachmentChunkingConfig;
import dev.fitko.fitconnect.api.domain.model.attachment.Attachment;
import dev.fitko.fitconnect.api.domain.model.event.Event;
import dev.fitko.fitconnect.api.domain.model.event.Status;
import dev.fitko.fitconnect.api.domain.model.metadata.AuthenticationInformation;
import dev.fitko.fitconnect.api.domain.model.metadata.ContentStructure;
import dev.fitko.fitconnect.api.domain.model.metadata.Hash;
import dev.fitko.fitconnect.api.domain.model.metadata.Metadata;
import dev.fitko.fitconnect.api.domain.model.metadata.SignatureType;
import dev.fitko.fitconnect.api.domain.model.metadata.attachment.ApiAttachment;
import dev.fitko.fitconnect.api.domain.model.metadata.attachment.Purpose;
import dev.fitko.fitconnect.api.domain.model.metadata.data.Data;
import dev.fitko.fitconnect.api.domain.model.metadata.data.MimeType;
import dev.fitko.fitconnect.api.domain.model.metadata.data.SubmissionSchema;
import dev.fitko.fitconnect.api.domain.model.reply.replychannel.ReplyChannel;
import dev.fitko.fitconnect.api.domain.sender.SendableEncryptedSubmission;
import dev.fitko.fitconnect.api.domain.sender.SendableSubmission;
import dev.fitko.fitconnect.api.domain.subscriber.ReceivedSubmission;
import dev.fitko.fitconnect.api.exceptions.client.FitConnectSenderException;
import dev.fitko.fitconnect.api.services.crypto.CryptoService;
import dev.fitko.fitconnect.client.SenderClient;
import dev.fitko.fitconnect.client.SubscriberClient;
import dev.fitko.fitconnect.client.bootstrap.ClientFactory;
import dev.fitko.fitconnect.client.util.ReplyChannelKeyGenerator;
import dev.fitko.fitconnect.core.crypto.HashService;
import dev.fitko.fitconnect.core.crypto.JWECryptoService;
import dev.fitko.fitconnect.integrationtests.condition.EnableIfEnvironmentVariablesAreSet;
import org.hamcrest.Matchers;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.text.ParseException;
import java.time.LocalDate;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import static dev.fitko.fitconnect.client.util.ResourceLoadingUtils.readResourceToString;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.isOneOf;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

@EnableIfEnvironmentVariablesAreSet
public class SenderClientIT extends IntegrationTestBase {

    @BeforeEach
    public void cleanup() {
        cleanupTestSubmissions();
    }

    @Test
    void testSendAndConfirmCycleWithJsonData() {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment();

        final String submissionData = readResourceToString("/submission_data.json");

        final var submission = SendableSubmission.Builder()
                .setDestination(UUID.fromString(System.getenv("TEST_DESTINATION_ID")))
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(submissionData, URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .addAttachment(Attachment.fromPath(Path.of("src/test/resources/attachment.txt"), "text/plain"))
                .addAttachment(Attachment.fromByteArray("attachment data".getBytes(), "text/plain"))
                .setReplyChannel(ReplyChannel.ofDeMail("test@mail.org"))
                .setCallback(URI.create("https://example.callback.net"), "callbackSecretWithMoreThan32Chars")
                .setAuthenticationInformation(List.of(new AuthenticationInformation("identificationReport", "1.0.0", "content")))
                .build();

        final var sentSubmission = ClientFactory.createSenderClient(config).send(submission);

        assertNotNull(sentSubmission);

        // When
        final ReceivedSubmission receivedSubmission =
                ClientFactory.createSubscriberClient(config)
                        .requestSubmission(sentSubmission.getSubmissionId());

        // Then
        assertNotNull(receivedSubmission);
        assertThat(receivedSubmission.getDataAsString(), is(submissionData));
        assertThat(receivedSubmission.getDataSchemaUri(), is(URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json")));
        assertThat(receivedSubmission.getDataMimeType(), is("application/json"));
        assertThat(receivedSubmission.getServiceType().getIdentifier(), is("urn:de:fim:leika:leistung:99400048079000"));
        assertThat(receivedSubmission.getAttachments(), hasSize(2));
        assertThat(receivedSubmission.getMetadata().getReplyChannel(), is(ReplyChannel.ofDeMail("test@mail.org")));
        assertThat(receivedSubmission.getAttachments().get(0).getDataAsString(), is("Test attachment"));
        assertThat(receivedSubmission.getAttachments().get(0).getMimeType(), is("text/plain"));
    }

    @Test
    void testSendAndConfirmCycleWithXmlData() throws IOException {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment();

        final String submissionData = readResourceToString("/submission_data.xml");

        final var submission = SendableSubmission.Builder()
                .setDestination(UUID.fromString(System.getenv("TEST_DESTINATION_ID")))
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setXmlData(submissionData, URI.create("urn:de:fim:leika:leistung:99400048079000"))
                .addAttachment(Attachment.fromPath(Path.of("src/test/resources/attachment.txt"), "text/plain"))
                .addAttachment(Attachment.fromByteArray("attachment data".getBytes(), "text/plain"))
                .setReplyChannel(ReplyChannel.ofDeMail("test@mail.org"))
                .setCallback(URI.create("https://example.callback.net"), "callBackSecretWithMoreThan32Chars")
                .build();

        final var sentSubmission = ClientFactory.createSenderClient(config).send(submission);

        assertNotNull(sentSubmission);

        // When
        final ReceivedSubmission receivedSubmission =
                ClientFactory.createSubscriberClient(config)
                        .requestSubmission(sentSubmission.getSubmissionId());

        // Then
        assertNotNull(receivedSubmission);
        assertThat(receivedSubmission.getDataAsString(), is(submissionData));
        assertThat(receivedSubmission.getDataSchemaUri(), is(URI.create("urn:de:fim:leika:leistung:99400048079000")));
        assertThat(receivedSubmission.getDataMimeType(), is("application/xml"));
        assertThat(receivedSubmission.getAttachments(), hasSize(2));
        assertThat(receivedSubmission.getMetadata().getReplyChannel(), is(ReplyChannel.ofDeMail("test@mail.org")));
        assertThat(new String(receivedSubmission.getAttachments().get(0).getDataAsBytes()), is("Test attachment"));
    }

    @Test
    void testNewSubmissionForExistingCaseWithFitConnectReplyChannel() {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment();

        final SenderClient senderClient = ClientFactory.createSenderClient(config);
        final SubscriberClient subscriberClient = ClientFactory.createSubscriberClient(config);

        final UUID destinationId = UUID.fromString(System.getenv("TEST_DESTINATION_ID"));
        final String submissionData = readResourceToString("/submission_data.json");

        final var firstSubmission = SendableSubmission.Builder()
                .setDestination(destinationId)
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(submissionData, URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .addAttachment(Attachment.fromPath(Path.of("src/test/resources/attachment.txt"), "text/plain"))
                .build();

        // When
        final var sentFirstSubmission = senderClient.send(firstSubmission);
        assertNotNull(sentFirstSubmission);

        final var secondSubmissionForCase = SendableSubmission.Builder()
                .setDestination(destinationId)
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(submissionData, URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .setCase(sentFirstSubmission.getCaseId())
                .setReplyChannel(ReplyChannel.ofFitConnect(ReplyChannelKeyGenerator.generateKeyPair().getPublicKey(), List.of("https://test.fitko.de")))
                .build();

        final var sentSecondSubmission = senderClient.send(secondSubmissionForCase);

        // Then
        assertNotNull(secondSubmissionForCase);
        assertThat(sentSecondSubmission.getCaseId(), is(sentFirstSubmission.getCaseId()));

        // check if reply channel communication is possible by getting the encryption key
        final ReceivedSubmission receivedSubmission = subscriberClient.requestSubmission(sentSecondSubmission.getSubmissionId());
        final JWK encryptionPublicKey = receivedSubmission.getMetadata().getReplyChannel().getFitConnect().getEncryptionPublicKey().toJwk();
        assertThat(encryptionPublicKey, is(secondSubmissionForCase.getReplyChannel().getFitConnect().getEncryptionPublicKey().toJwk()));
    }


    @Test
    void testSendAndConfirmCycleWithEncryptedData() throws ParseException, IOException {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment();
        final UUID destinationId = UUID.fromString(System.getenv("TEST_DESTINATION_ID"));

        final CryptoService cryptoService = new JWECryptoService(new HashService());
        final SenderClient senderClient = ClientFactory.createSenderClient(config);
        final SubscriberClient subscriberClient = ClientFactory.createSubscriberClient(config);

        final RSAKey encryptionKey = RSAKey.parse(senderClient.getPublicKeyForDestination(destinationId));

        final String jsonData = readResourceToString("/submission_data.json");
        final File attachmentFile = new File("src/test/resources/attachment.txt");
        final String attachmentData = Files.readString(attachmentFile.toPath());

        final String encryptedData = cryptoService.encryptBytes(encryptionKey, jsonData.getBytes(StandardCharsets.UTF_8), "application/json");
        final String encryptedAttachment = cryptoService.encryptBytes(encryptionKey, attachmentData.getBytes(StandardCharsets.UTF_8), "text/plain");

        final var submissionSchema = new SubmissionSchema();
        submissionSchema.setSchemaUri(URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"));
        submissionSchema.setMimeType(MimeType.APPLICATION_JSON);

        final var dataHash = new Hash();
        dataHash.setSignatureType(SignatureType.SHA_512);
        dataHash.setContent(cryptoService.hashBytes(jsonData.getBytes(StandardCharsets.UTF_8)));

        final var data = new Data();
        data.setHash(dataHash);
        data.setSubmissionSchema(submissionSchema);

        final var attachment = new ApiAttachment();
        attachment.setAttachmentId(UUID.randomUUID());
        attachment.setPurpose(Purpose.ATTACHMENT);
        attachment.setFilename(attachmentFile.getName());
        attachment.setMimeType("text/plain");

        final var attachmentHash = new Hash();
        attachmentHash.setContent(cryptoService.hashBytes(attachmentData.getBytes(StandardCharsets.UTF_8)));
        attachmentHash.setSignatureType(SignatureType.SHA_512);
        attachment.setHash(attachmentHash);

        final var contentStructure = new ContentStructure();
        contentStructure.setAttachments(List.of(attachment));
        contentStructure.setData(data);

        final var metadata = new Metadata();
        metadata.setSchema(config.getMetadataSchemaWriteVersion().toString());
        metadata.setContentStructure(contentStructure);

        final String encryptedMetadata = cryptoService.encryptObject(encryptionKey, metadata, "application/json");

        // When
        final var submission = SendableEncryptedSubmission.Builder()
                .setDestination(destinationId)
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setEncryptedMetadata(encryptedMetadata)
                .setEncryptedData(encryptedData)
                .addEncryptedAttachment(attachment.getAttachmentId(), encryptedAttachment)
                .build();

        final var sentSubmission = senderClient.send(submission);

        assertNotNull(sentSubmission);

        final ReceivedSubmission receivedSubmission = subscriberClient.requestSubmission(sentSubmission.getSubmissionId());

        // Then
        assertNotNull(receivedSubmission);
        assertThat(receivedSubmission.getDataAsString(), Matchers.is(jsonData));
        assertThat(receivedSubmission.getDataMimeType(), Matchers.is("application/json"));
        assertThat(receivedSubmission.getAttachments().get(0).getDataAsString(StandardCharsets.UTF_8), Matchers.is("Test attachment"));
    }

    @Test
    void testNewEncryptedSubmissionForExistingCaseWithFitConnectReplyChannel() throws ParseException, IOException {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment();
        final UUID destinationId = UUID.fromString(System.getenv("TEST_DESTINATION_ID"));

        final CryptoService cryptoService = new JWECryptoService(new HashService());
        final SenderClient senderClient = ClientFactory.createSenderClient(config);
        final SubscriberClient subscriberClient = ClientFactory.createSubscriberClient(config);

        final RSAKey encryptionKey = RSAKey.parse(senderClient.getPublicKeyForDestination(destinationId));

        final String jsonData = readResourceToString("/submission_data.json");
        final File attachmentFile = new File("src/test/resources/attachment.txt");
        final String attachmentData = Files.readString(attachmentFile.toPath());

        final String encryptedData = cryptoService.encryptBytes(encryptionKey, jsonData.getBytes(StandardCharsets.UTF_8), MimeType.APPLICATION_JSON.value());
        final String encryptedAttachment = cryptoService.encryptBytes(encryptionKey, attachmentData.getBytes(StandardCharsets.UTF_8), "text/plain");

        final var submissionSchema = new SubmissionSchema();
        submissionSchema.setSchemaUri(URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"));
        submissionSchema.setMimeType(MimeType.APPLICATION_JSON);

        final var dataHash = new Hash();
        dataHash.setSignatureType(SignatureType.SHA_512);
        dataHash.setContent(cryptoService.hashBytes(jsonData.getBytes(StandardCharsets.UTF_8)));

        final var data = new Data();
        data.setHash(dataHash);
        data.setSubmissionSchema(submissionSchema);

        final var attachment = new ApiAttachment();
        attachment.setAttachmentId(UUID.randomUUID());
        attachment.setPurpose(Purpose.ATTACHMENT);
        attachment.setFilename(attachmentFile.getName());
        attachment.setMimeType("text/plain");

        final var attachmentHash = new Hash();
        attachmentHash.setContent(cryptoService.hashBytes(attachmentData.getBytes(StandardCharsets.UTF_8)));
        attachmentHash.setSignatureType(SignatureType.SHA_512);
        attachment.setHash(attachmentHash);

        final var contentStructure = new ContentStructure();
        contentStructure.setAttachments(List.of(attachment));
        contentStructure.setData(data);

        final var metadata = new Metadata();
        metadata.setSchema(config.getMetadataSchemaWriteVersion().toString());
        metadata.setContentStructure(contentStructure);

        final var contentStructureForNewSubmission = new ContentStructure();
        contentStructureForNewSubmission.setData(data);

        final var metadataForNewSubmission = new Metadata();
        metadataForNewSubmission.setSchema(config.getMetadataSchemaWriteVersion().toString());
        metadataForNewSubmission.setContentStructure(contentStructureForNewSubmission);
        metadataForNewSubmission.setReplyChannel(ReplyChannel.ofFitConnect(ReplyChannelKeyGenerator.generateKeyPair().getPublicKey(), List.of("https://test.fitko.de")));

        final String encryptedMetadata = cryptoService.encryptBytes(encryptionKey, new ObjectMapper().writeValueAsBytes(metadata), MimeType.APPLICATION_JSON.value());
        final String encryptedMetadataForNewSubmission = cryptoService.encryptBytes(encryptionKey, new ObjectMapper().writeValueAsBytes(metadataForNewSubmission), MimeType.APPLICATION_JSON.value());

        // When
        final var initialSubmission = SendableEncryptedSubmission.Builder()
                .setDestination(destinationId)
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setEncryptedMetadata(encryptedMetadata)
                .setEncryptedData(encryptedData)
                .addEncryptedAttachment(attachment.getAttachmentId(), encryptedAttachment)
                .build();

        final var sentInitialSubmission = senderClient.send(initialSubmission);
        assertNotNull(sentInitialSubmission);

        final var secondSubmissionForExistingCase = SendableEncryptedSubmission.Builder()
                .setDestination(destinationId)
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setEncryptedMetadata(encryptedMetadataForNewSubmission)
                .setEncryptedData(encryptedData)
                .setCase(sentInitialSubmission.getCaseId())
                .build();

        // Then
        final var sentSecondSubmissionForExistingCase = senderClient.send(secondSubmissionForExistingCase);
        assertNotNull(sentSecondSubmissionForExistingCase);

        final ReceivedSubmission receivedNewSubmissionForCase = subscriberClient.requestSubmission(sentSecondSubmissionForExistingCase.getSubmissionId());
        assertNotNull(sentSecondSubmissionForExistingCase);

        final JWK encryptionPublicKey = receivedNewSubmissionForCase.getMetadata().getReplyChannel().getFitConnect().getEncryptionPublicKey().toJwk();
        assertThat(encryptionPublicKey, is(metadataForNewSubmission.getReplyChannel().getFitConnect().getEncryptionPublicKey().toJwk()));
    }

    @Test
    void testSubmissionDataValidationFromLocalSchemaFile(@TempDir final Path tempDir) throws IOException {

        // Given

        // Prepare locally set submission data schema in given path
        final String schemaResource = readResourceToString("/submission-data-schema/submission_data_schema.json");
        final Path schemaPath = Files.writeString(Path.of(tempDir.toString(), "submission_data_schema.json"), schemaResource);
        final String schemaUri = "https://schema.fitko.de/fim/s00000114_1.1.schema.json";

        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment(true, Map.of(schemaUri, schemaPath.toString()));

        final String submissionData = readResourceToString("/submission_data.json");
        final URI submissionDataSchemaUri = URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json");

        final var submission = SendableSubmission.Builder()
                .setDestination(UUID.fromString(System.getenv("TEST_DESTINATION_ID")))
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(submissionData, submissionDataSchemaUri)
                .addAttachment(Attachment.fromPath(Path.of("src/test/resources/attachment.txt"), "text/plain"))
                .addAttachment(Attachment.fromByteArray("attachment data".getBytes(), "text/plain"))
                .setReplyChannel(ReplyChannel.ofDeMail("test@mail.org"))
                .build();

        // When
        final var sentSubmission = ClientFactory.createSenderClient(config).send(submission);

        // Then
        assertNotNull(sentSubmission);
    }

    @Test
    void testAbortedSendSubmissionWithKeyValidationNotSilent() {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment(false);

        // When
        final var submission = SendableSubmission.Builder()
                .setDestination(UUID.fromString(System.getenv("TEST_DESTINATION_ID")))
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(readResourceToString("/submission_data.json"), URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .build();

        final FitConnectSenderException exception = assertThrows(FitConnectSenderException.class, () -> ClientFactory.createSenderClient(config).send(submission));

        // Then
        assertThat(exception.getMessage(), containsString("Invalid public encryption key"));
        assertThat(exception.getCause().getMessage(), containsString("JWK with id jsMQHFiA2uMGtiG3Ro8RJnYdEhp5W5KjGW-Vcf3-YMk has an invalid amount of certificates. Is 1 but should be 3."));
    }

    @Test
    void testReadSubmissionStatus() {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment();

        final String submissionData = readResourceToString("/submission_data.json");

        final SenderClient senderClient = ClientFactory.createSenderClient(config);

        final var submission = SendableSubmission.Builder()
                .setDestination(UUID.fromString(System.getenv("TEST_DESTINATION_ID")))
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(submissionData, URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .addAttachment(Attachment.fromPath(Path.of("src/test/resources/attachment.txt"), "plain/text"))
                .build();

        final var sentSubmission = senderClient.send(submission);

        assertNotNull(sentSubmission);

        final Status statusForSubmission = senderClient.getSubmissionStatus(sentSubmission);

        assertThat(statusForSubmission.getState(), isOneOf(Event.SUBMIT_SUBMISSION.getState(), Event.NOTIFY_SUBMISSION.getState()));
        assertThat(statusForSubmission.getIssuer(), containsString("https://test.fit-connect.fitko.dev"));
    }

    @Test
    void testSendSubmissionWithAdditionalReferenceInfo() {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment(true);

        final String submissionData = readResourceToString("/submission_data.json");

        final LocalDate applicationDate = LocalDate.of(2020, 1, 1);
        final UUID idBundDeApplicationId = UUID.randomUUID();

        final var submission = SendableSubmission.Builder()
                .setDestination(UUID.fromString(System.getenv("TEST_DESTINATION_ID")))
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(submissionData, URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .setApplicationDate(applicationDate)
                .setIdBundDeApplicationId(idBundDeApplicationId)
                .build();

        final var sentSubmission = ClientFactory.createSenderClient(config).send(submission);

        assertNotNull(sentSubmission);

        // When
        final ReceivedSubmission receivedSubmission =
                ClientFactory.createSubscriberClient(config)
                        .requestSubmission(sentSubmission.getSubmissionId());

        // Then
        assertNotNull(receivedSubmission);
        assertTrue(receivedSubmission.getApplicationDate().isPresent());
        assertThat(receivedSubmission.getApplicationDate().get(), is(applicationDate));
        assertThat(receivedSubmission.getMetadata().getAdditionalReferenceInfo().getIdBundDeApplicationId(), is(idBundDeApplicationId));
    }

    @Test
    void testSendSubmissionWithAdditionalReferenceInfoAndBundIdReplyChannel() {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment(true);
        final String submissionData = readResourceToString("/submission_data.json");

        final UUID idBundDeApplicationId = UUID.randomUUID();
        final UUID mailboxId = UUID.randomUUID();

        final var submission = SendableSubmission.Builder()
                .setDestination(UUID.fromString(System.getenv("TEST_DESTINATION_ID")))
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(submissionData, URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .setReplyChannel(ReplyChannel.ofIdBundDeMailbox(mailboxId))
                .setIdBundDeApplicationId(idBundDeApplicationId)
                .build();

        final var sentSubmission = ClientFactory.createSenderClient(config).send(submission);

        assertNotNull(sentSubmission);

        // When
        final ReceivedSubmission receivedSubmission =
                ClientFactory.createSubscriberClient(config)
                        .requestSubmission(sentSubmission.getSubmissionId());

        // Then
        assertNotNull(receivedSubmission.getMetadata().getReplyChannel());
        assertNotNull(receivedSubmission.getMetadata().getAdditionalReferenceInfo());
        assertTrue(receivedSubmission.getMetadata().getReplyChannel().isIdBundDeMailbox());
        assertThat(receivedSubmission.getMetadata().getReplyChannel().getIdBundDeMailbox().getMailboxUuid(), is(mailboxId));
        assertThat(receivedSubmission.getMetadata().getAdditionalReferenceInfo().getIdBundDeApplicationId(), is(idBundDeApplicationId));
    }

    @Test
    void testSendSubmissionWithChunkedAttachments(@TempDir Path attachmentStoragePath) throws IOException {

        // Given
        final AttachmentChunkingConfig chunkingConfig = AttachmentChunkingConfig.builder()
                .chunkAllAttachments(true)
                .chunkSizeInMB(1)
                .attachmentStoragePath(attachmentStoragePath)
                .build();

        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment(true, chunkingConfig);

        final SenderClient senderClient = ClientFactory.createSenderClient(config);

        final Path attachmentPath = Path.of("src/test/resources/large-pdf.pdf");
        final byte[] expectedAttachmentData = Files.readAllBytes(attachmentPath);

        final var submission = SendableSubmission.Builder()
                .setDestination(UUID.fromString(System.getenv("TEST_DESTINATION_ID")))
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(readResourceToString("/submission_data.json"), URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .addAttachment(Attachment.fromLargeAttachment(attachmentPath, "application/pdf", "large-pdf_1.pdf", "test large pdf"))
                .addAttachment(Attachment.fromPath(attachmentPath, "application/pdf", "large-pdf_3.pdf", "test in-memory pdf"))
                .build();

        final var sentSubmission = senderClient.send(submission);

        assertNotNull(sentSubmission);

        // When
        final ReceivedSubmission receivedSubmission =
                ClientFactory.createSubscriberClient(config)
                        .requestSubmission(sentSubmission.getSubmissionId());

        // Then
        assertNotNull(receivedSubmission);
        assertThat(receivedSubmission.getAttachments(), hasSize(2));
        assertArrayEquals(expectedAttachmentData, receivedSubmission.getAttachments().get(0).getDataAsBytes());
        assertArrayEquals(expectedAttachmentData, receivedSubmission.getAttachments().get(1).getDataAsBytes());

        // chunking all attachments will lead to attachments that are all stored in filesystem
        assertTrue(receivedSubmission.getAttachments().get(0).isLargeAttachment());
        assertTrue(receivedSubmission.getAttachments().get(1).isLargeAttachment());

        senderClient.clearAttachmentStorage();
        assertThat(attachmentStoragePath.toFile().listFiles().length, is(0));
    }

    @Test
    void testSendSubmissionWithDefaultChunkingForLargeAttachmentsOnly() throws IOException {

        // Given
        final AttachmentChunkingConfig chunkingConfig = AttachmentChunkingConfig.builder()
                .chunkAllAttachments(false)
                .chunkSizeInMB(1)
                .build();

        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment(true, chunkingConfig);

        final Path attachmentPath = Path.of("src/test/resources/large-pdf.pdf");
        final byte[] expectedAttachmentData = Files.readAllBytes(attachmentPath);

        final var submission = SendableSubmission.Builder()
                .setDestination(UUID.fromString(System.getenv("TEST_DESTINATION_ID")))
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(readResourceToString("/submission_data.json"), URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .addAttachment(Attachment.fromLargeAttachment(attachmentPath, "application/pdf", "large-pdf_1.pdf", "test large pdf"))
                .addAttachment(Attachment.fromPath(attachmentPath, "application/pdf", "large-pdf_3.pdf", "test in-memory pdf"))
                .build();

        final var sentSubmission = ClientFactory.createSenderClient(config).send(submission);

        assertNotNull(sentSubmission);

        // When
        final ReceivedSubmission receivedSubmission =
                ClientFactory.createSubscriberClient(config)
                        .requestSubmission(sentSubmission.getSubmissionId());

        // Then
        assertNotNull(receivedSubmission);
        assertThat(receivedSubmission.getAttachments(), hasSize(2));
        assertArrayEquals(expectedAttachmentData, receivedSubmission.getAttachments().get(0).getDataAsBytes());
        assertArrayEquals(expectedAttachmentData, receivedSubmission.getAttachments().get(1).getDataAsBytes());

        assertTrue(receivedSubmission.getAttachments().get(0).isLargeAttachment());
        assertTrue(receivedSubmission.getAttachments().get(1).isInMemoryAttachment());
    }

    @Test
    void testMultiThreadExecution() throws InterruptedException {

        // Given
        final ApplicationConfig config = getConfigWithCredentialsFromEnvironment();
        final SenderClient senderClient = ClientFactory.createSenderClient(config);
        final SubscriberClient subscriberClient = ClientFactory.createSubscriberClient(config);

        final String submissionData = readResourceToString("/submission_data.json");
        final UUID destinationId = UUID.fromString(System.getenv("TEST_DESTINATION_ID"));
        final var submission = SendableSubmission.Builder()
                .setDestination(destinationId)
                .setServiceType("urn:de:fim:leika:leistung:99400048079000", "Test Service")
                .setJsonData(submissionData, URI.create("https://schema.fitko.de/fim/s00000114_1.1.schema.json"))
                .addAttachment(Attachment.fromPath(Path.of("src/test/resources/attachment.txt"), "plain/text"))
                .build();

        // When
        final ExecutorService es = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 25; i++) {
            es.execute(new SenderRunner(senderClient, submission, "Thread-" + i));
        }
        es.shutdown();
        es.awaitTermination(1, TimeUnit.MINUTES);

        // wait for ZSD processing
        Thread.sleep(5000);

        final int handedInSubmissions = subscriberClient.getAvailableSubmissionsForDestination(destinationId).size();

        // Then
        assertThat(handedInSubmissions, is(25));
    }

    static class SenderRunner implements Runnable {
        private final SenderClient senderClient;
        private final SendableSubmission submission;
        private final String threadName;

        SenderRunner(final SenderClient senderClient, final SendableSubmission submission, final String threadName) {
            this.senderClient = senderClient;
            this.submission = submission;
            this.threadName = threadName;
            System.out.println("Creating " + threadName);
        }

        public void run() {
            System.out.println("Running " + threadName);
            assertNotNull(senderClient.send(submission));
            System.out.println("Thread " + threadName + " exiting.");
        }
    }

}