/*
 * Copyright 2020 Saxon State and University Library Dresden (SLUB)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package de.qucosa.metsdisseminator.application;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import de.qucosa.metsdisseminator.events.ActivityStreamsEventType;
import de.qucosa.metsdisseminator.events.FedoraUpdateEvent;
import de.qucosa.metsdisseminator.events.ProcessResultEvent;
import de.qucosa.metsdisseminator.repository.fcrepo3.Namespaces;
import de.qucosa.testsupport.Fcrepo3Container;
import de.qucosa.testsupport.FcrepoContainer;
import de.qucosa.testsupport.KafkaConsumerRule;
import de.qucosa.testsupport.KafkaProducerRule;
import org.apache.camel.component.properties.PropertiesComponent;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.apache.commons.io.IOUtils;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.custommonkey.xmlunit.SimpleNamespaceContext;
import org.custommonkey.xmlunit.XMLUnit;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.testcontainers.containers.KafkaContainer;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.Charset;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Iterator;
import java.util.Optional;
import java.util.Properties;
import java.util.UUID;

import static de.qucosa.testsupport.KafkaConsumerRule.OffsetReset.EARLIEST;
import static java.util.Collections.singletonList;

public class IntegrationTestSupport extends CamelTestSupport {

    @ClassRule(order = 1)
    public final static KafkaContainer kafkaContainer = new KafkaContainer();
    @ClassRule(order = 1)
    public final static Fcrepo3Container fcrepo3Container = new Fcrepo3Container();
    @ClassRule(order = 1)
    public final static FcrepoContainer fcrepoContainer = new FcrepoContainer();
    @ClassRule(order = 2)
    public final static KafkaProducerRule kafkaProducer = new KafkaProducerRule(kafkaContainer);
    @ClassRule(order = 2)
    public final static KafkaConsumerRule kafkaConsumer = new KafkaConsumerRule(kafkaContainer, EARLIEST);
    private static final URI FEDORA3_SOURCE_ID = URI.create("urn:fcrepo3:container");
    private ObjectMapper objectMapper;

    @Override
    public boolean isUseAdviceWith() {
        return true;
    }

    @Before
    public void contextSetup() throws Exception {
        final PropertiesComponent props = new PropertiesComponent();
        props.setLocation("classpath:default.properties");

        overridePropertiesSetup(props);

        context.setPropertiesComponent(props);

        context.addLifecycleStrategy(new DataFormatsContextConfiguration());
        context.addLifecycleStrategy(new KafkaContextConfiguration());
        context.addLifecycleStrategy(new Fedora3ContextConfiguration());

        adviceRoutes();

        startCamelContext();

        objectMapper = context.getRegistry().lookupByNameAndType("objectmapper", ObjectMapper.class);
    }

    @BeforeClass
    static public void setupXpathNamespaces() {
        XMLUnit.setXpathNamespaceContext(new SimpleNamespaceContext(Namespaces.getPrefixUriMap()));
    }

    @SuppressWarnings("EmptyMethod")
    protected void adviceRoutes() throws Exception {
        // NOOP
    }

    @SuppressWarnings("EmptyMethod")
    protected void overrideProperties(Properties properties) {
        // NOOP
    }

    protected String getContentAndClose(CloseableHttpResponse response) throws IOException {
        final String contentString = IOUtils.toString(
                response.getEntity().getContent().readAllBytes(),
                Charset.defaultCharset().name());
        response.close();
        return contentString;
    }

    protected ProcessResultEvent receiveProcessResultEvent() {
        final ConsumerRecords<String, String> processResultEvents =
                kafkaConsumer.subscribeAndPoll(
                        Duration.ofSeconds(15),
                        "process_results");
        final Optional<ConsumerRecord<String, String>> record =
                firstOf(processResultEvents.records("process_results")
                        .iterator());
        try {
            return (record.isPresent()) ? objectMapper.readValue(record.get().value(), ProcessResultEvent.class) : null;
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    protected void issueServiceEvent(URI identifier, ActivityStreamsEventType eventType) {
        try {
            kafkaProducer.send("service_events",
                    objectMapper.writeValueAsString(
                            new FedoraUpdateEvent(
                                    FEDORA3_SOURCE_ID,
                                    fcrepo3Container.apiUri().toURL(),
                                    UUID.randomUUID(),
                                    singletonList(eventType),
                                    identifier,
                                    singletonList(URI.create("https://www.w3.org/ns/ldp#Container")),
                                    LocalDateTime.now(),
                                    "anja")));
        } catch (JsonProcessingException | MalformedURLException e) {
            throw new RuntimeException(e);
        }
    }

    private void overridePropertiesSetup(PropertiesComponent props) throws URISyntaxException {
        final Properties overrideProperties = new Properties();
        overrideProperties.setProperty("kafka.broker", kafkaContainer.getBootstrapServers());

        overrideProperties.setProperty("fcrepo3.sources", "f1");
        overrideProperties.setProperty("fcrepo3.default", "f1");
        overrideProperties.setProperty("f1.sourceID", FEDORA3_SOURCE_ID.toString());
        overrideProperties.setProperty("f1.location", fcrepo3Container.apiUri().toString());
        overrideProperties.setProperty("f1.user", fcrepo3Container.username());
        overrideProperties.setProperty("f1.password", fcrepo3Container.password());

        overrideProperties.setProperty("fedora.host", fcrepoContainer.apiUri().toString());
        overrideProperties.setProperty("fedora.user", fcrepoContainer.username());
        overrideProperties.setProperty("fedora.password", fcrepoContainer.password());

        overrideProperties(overrideProperties);

        props.setOverrideProperties(overrideProperties);
    }

    private <T> Optional<T> firstOf(Iterator<T> iterator) {
        return iterator.hasNext() ? Optional.of(iterator.next()) : Optional.empty();
    }
}
