/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.camel.quarkus.component.dataset.it;

import java.nio.charset.StandardCharsets;
import java.nio.file.Files;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.Response;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.dataset.DataSetEndpoint;
import org.apache.camel.component.dataset.DataSetTestEndpoint;
import org.apache.camel.component.dataset.FileDataSet;
import org.apache.camel.component.dataset.ListDataSet;
import org.apache.camel.component.dataset.SimpleDataSet;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.support.CamelContextHelper;

@Path("/dataset")
@ApplicationScoped
public class DataSetResource {
    @Inject
    ProducerTemplate producerTemplate;

    @Inject
    CamelContext context;

    @GET
    @Path("/simple")
    public void dataSetSimple(@QueryParam("useIndexHeader") boolean useIndexHeader) throws InterruptedException {
        String dataSetName = useIndexHeader ? "simpleDataSetWithIndex" : "simpleDataSet";
        SimpleDataSet simpleDataSet = CamelContextHelper.lookup(context, dataSetName, SimpleDataSet.class);

        DataSetEndpoint endpoint = context.getEndpoint("dataset:" + dataSetName + "?minRate=50", DataSetEndpoint.class);
        endpoint.expectedMessageCount((int) simpleDataSet.getSize());

        for (int i = 0; i < simpleDataSet.getSize(); i++) {
            if (useIndexHeader) {
                producerTemplate.sendBodyAndHeader("direct:simpleDataSetWithIndex", simpleDataSet.getDefaultBody(),
                        Exchange.DATASET_INDEX, i);
            } else {
                producerTemplate.sendBody("direct:simpleDataSet", simpleDataSet.getDefaultBody());
            }
        }

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            endpoint.reset();
        }
    }

    @GET
    @Path("/simple/exception")
    public Response simpleDataSetAssertionException() throws InterruptedException {
        DataSetEndpoint endpoint = context.getEndpoint("dataset:simpleDataSetForException", DataSetEndpoint.class);

        producerTemplate.sendBody("direct:simpleDataSetForException", "invalid payload");

        try {
            endpoint.assertIsSatisfied(5000);
        } catch (Throwable t) {
            if (t instanceof AssertionError) {
                return Response.ok().build();
            } else {
                return Response.serverError().build();
            }
        } finally {
            endpoint.reset();
        }

        return Response.serverError().build();
    }

    @GET
    @Path("/simple/consumer")
    public void dataSetSimpleConsumer() throws Exception {
        SimpleDataSet simpleDataSet = CamelContextHelper.lookup(context, "simpleDataSetForConsumer", SimpleDataSet.class);
        MockEndpoint endpoint = context.getEndpoint("mock:simpleDataSetResult", MockEndpoint.class);
        endpoint.expectedMessageCount((int) simpleDataSet.getSize());
        endpoint.expectedMessagesMatches(exchange -> {
            String body = exchange.getMessage().getBody(String.class);
            return body != null && body.equals("<hello>world!</hello>");
        });

        context.getRouteController().startRoute("simple");
        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            context.getRouteController().stopRoute("simple");
            endpoint.reset();
        }
    }

    @GET
    @Path("/list")
    public void dataSetList() throws InterruptedException {
        ListDataSet listDataSet = CamelContextHelper.lookup(context, "listDataSet", ListDataSet.class);
        DataSetEndpoint endpoint = context.getEndpoint("dataset:listDataSet", DataSetEndpoint.class);
        endpoint.expectedMessageCount((int) listDataSet.getSize());

        producerTemplate.sendBodyAndHeader("direct:listDataSet", "Hello", Exchange.DATASET_INDEX, 0);
        producerTemplate.sendBodyAndHeader("direct:listDataSet", "World", Exchange.DATASET_INDEX, 1);

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            endpoint.reset();
        }
    }

    @GET
    @Path("/list/consumer")
    public void dataSetListConsumer() throws Exception {
        MockEndpoint endpoint = context.getEndpoint("mock:listDataSetResult", MockEndpoint.class);
        endpoint.expectedBodiesReceived("Hello", "World");

        context.getRouteController().startRoute("list");
        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            context.getRouteController().stopRoute("list");
            endpoint.reset();
        }
    }

    @GET
    @Path("/file")
    public void dataSetFile() throws InterruptedException {
        FileDataSet fileDataSet = CamelContextHelper.lookup(context, "fileDataSet", FileDataSet.class);
        DataSetEndpoint endpoint = context.getEndpoint("dataset:fileDataSet", DataSetEndpoint.class);
        endpoint.expectedMessageCount((int) fileDataSet.getSize());

        producerTemplate.sendBodyAndHeader("direct:fileDataSet", "Hello World\n", Exchange.DATASET_INDEX, 0);

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            endpoint.reset();
        }
    }

    @GET
    @Path("/file/delimited")
    public void dataSetFileDelimited() throws InterruptedException {
        FileDataSet fileDataSet = CamelContextHelper.lookup(context, "fileDataSetDelimited", FileDataSet.class);
        DataSetEndpoint endpoint = context.getEndpoint("dataset:fileDataSetDelimited", DataSetEndpoint.class);
        endpoint.expectedMessageCount((int) fileDataSet.getSize());

        producerTemplate.sendBodyAndHeader("direct:fileDataSetDelimited", "Hello", Exchange.DATASET_INDEX, 0);
        producerTemplate.sendBodyAndHeader("direct:fileDataSetDelimited", "World", Exchange.DATASET_INDEX, 1);

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            endpoint.reset();
        }
    }

    @GET
    @Path("/file/consumer")
    public void dataSetFileConsumer() throws Exception {
        MockEndpoint endpoint = context.getEndpoint("mock:fileDataSetResult", MockEndpoint.class);
        endpoint.expectedBodiesReceived("Hello World\n");

        context.getRouteController().startRoute("file");
        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            context.getRouteController().stopRoute("file");
            endpoint.reset();
        }
    }

    @GET
    @Path("/custom")
    public void dataSetCustom() throws Exception {
        context.getRouteController().startRoute("custom");

        CustomDataSet customDataSet = CamelContextHelper.lookup(context, "customDataSet", CustomDataSet.class);
        DataSetEndpoint endpoint = context.getEndpoint("dataset:customDataSet", DataSetEndpoint.class);
        endpoint.setExpectedMessageCount((int) customDataSet.getSize());

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            context.getRouteController().stopRoute("custom");
            endpoint.reset();
        }
    }

    @GET
    @Path("/simple/index/off")
    public void dataSetIndexOff() throws InterruptedException {
        SimpleDataSet simpleDataSet = CamelContextHelper.lookup(context, "simpleDataSetIndexOff", SimpleDataSet.class);
        DataSetEndpoint endpoint = context.getEndpoint("dataset:simpleDataSetIndexOff?dataSetIndex=off", DataSetEndpoint.class);
        endpoint.setExpectedMessageCount((int) simpleDataSet.getSize());

        for (int i = 0; i < simpleDataSet.getSize(); i++) {
            if (0 == i % 2) {
                producerTemplate.sendBodyAndHeader("direct:simpleDataSetIndexOff", simpleDataSet.getDefaultBody(),
                        Exchange.DATASET_INDEX, i - 1);
            } else {
                producerTemplate.sendBody("direct:simpleDataSetIndexOff", simpleDataSet.getDefaultBody());
            }
        }

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            endpoint.reset();
        }
    }

    @GET
    @Path("/simple/index/lenient")
    public void dataSetIndexLenient() throws InterruptedException {
        SimpleDataSet simpleDataSet = CamelContextHelper.lookup(context, "simpleDataSetIndexLenient", SimpleDataSet.class);
        DataSetEndpoint endpoint = context.getEndpoint("dataset:simpleDataSetIndexLenient?dataSetIndex=lenient",
                DataSetEndpoint.class);
        endpoint.setExpectedMessageCount((int) simpleDataSet.getSize());

        for (int i = 0; i < simpleDataSet.getSize(); i++) {
            if (0 == i % 2) {
                producerTemplate.sendBodyAndHeader("direct:simpleDataSetIndexLenient", simpleDataSet.getDefaultBody(),
                        Exchange.DATASET_INDEX, i);
            } else {
                producerTemplate.sendBody("direct:simpleDataSetIndexLenient", simpleDataSet.getDefaultBody());
            }
        }

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            endpoint.reset();
        }
    }

    @GET
    @Path("/simple/index/strict")
    public Response dataSetIndexStrict(@QueryParam("useIndexHeader") boolean useIndexHeader) throws InterruptedException {
        String dataSet = useIndexHeader ? "simpleDataSetIndexStrict" : "simpleDataSetIndexStrictWithoutHeader";
        SimpleDataSet simpleDataSet = CamelContextHelper.lookup(context, dataSet, SimpleDataSet.class);
        DataSetEndpoint endpoint = context.getEndpoint("dataset:" + dataSet + "?dataSetIndex=strict", DataSetEndpoint.class);
        endpoint.setExpectedMessageCount((int) simpleDataSet.getSize());

        for (int i = 0; i < simpleDataSet.getSize(); i++) {
            if (useIndexHeader) {
                producerTemplate.sendBodyAndHeader("direct:simpleDataSetIndexStrict", simpleDataSet.getDefaultBody(),
                        Exchange.DATASET_INDEX, i);
            } else {
                producerTemplate.sendBody("direct:simpleDataSetIndexStrictWithoutHeader", simpleDataSet.getDefaultBody());
            }
        }

        try {
            endpoint.assertIsSatisfied(5000);
        } catch (Throwable t) {
            return Response.serverError().entity(t.getMessage()).build();
        } finally {
            endpoint.reset();
        }

        return Response.ok().build();
    }

    @GET
    @Path("/preload")
    public void dataSetPreloaded() throws Exception {
        SimpleDataSet preloadedDataSet = CamelContextHelper.lookup(context, "preloadedDataSet", SimpleDataSet.class);
        DataSetEndpoint endpoint = context.getEndpoint("dataset:preloadedDataSet?preloadSize=5", DataSetEndpoint.class);
        endpoint.setExpectedMessageCount((int) preloadedDataSet.getSize());

        context.getRouteController().startRoute("preload");
        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            context.getRouteController().stopRoute("preload");
            endpoint.reset();
        }

    }

    @GET
    @Path("/test/seda")
    public void dataSetTestSeda() throws Exception {
        producerTemplate.sendBody("seda:dataSetTestSeda", "Hello World");

        // We do this here so that dataset-test can pull the messages sent to the seda endpoint
        context.addRoutes(new RouteBuilder() {
            public void configure() throws Exception {
                from("direct:dataSetTestSeda").id("dataSetTestSeda")
                        .to("dataset-test:seda:dataSetTestSeda?timeout=0");
            }
        });

        DataSetTestEndpoint endpoint = context.getEndpoint("dataset-test:seda:dataSetTestSeda?timeout=0",
                DataSetTestEndpoint.class);
        endpoint.expectedMessageCount(1);

        producerTemplate.sendBody("direct:dataSetTestSeda", "Hello World");

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            context.removeRoute("dataSetTestSeda");
            endpoint.reset();
        }
    }

    @GET
    @Path("/test/seda/any/order")
    public void dataSetTestSedaAnyOrder() throws Exception {
        producerTemplate.sendBody("seda:dataSetTestSedaAnyOrder", "Hello World");
        producerTemplate.sendBody("seda:dataSetTestSedaAnyOrder", "Bye World");

        // We do this here so that dataset-test can pull the messages sent to the seda endpoint
        context.addRoutes(new RouteBuilder() {
            public void configure() throws Exception {
                from("direct:dataSetTestSedaAnyOrder").id("dataSetTestSedaAnyOrder")
                        .to("dataset-test:seda:dataSetTestSedaAnyOrder?timeout=0&anyOrder=true");
            }
        });

        DataSetTestEndpoint endpoint = context.getEndpoint("dataset-test:seda:dataSetTestSedaAnyOrder?timeout=0&anyOrder=true",
                DataSetTestEndpoint.class);
        endpoint.expectedMessageCount(2);

        producerTemplate.sendBody("direct:dataSetTestSedaAnyOrder", "Bye World");
        producerTemplate.sendBody("direct:dataSetTestSedaAnyOrder", "Hello World");

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            context.removeRoute("dataSetTestSedaAnyOrder");
            endpoint.reset();
        }
    }

    @GET
    @Path("/test/split")
    public void dataSetTestSplit() throws Exception {
        java.nio.file.Path path = Files.createTempFile("dataSetTestSplit", ".txt");
        Files.write(path, "Hello\nWorld\nBye\nWorld".getBytes(StandardCharsets.UTF_8));

        context.addRoutes(new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct:dataSetTestSplit").id("dataSetTestSplit")
                        .toF("dataset-test:file:%s?noop=true&split=true&timeout=1000&fileName=%s", path.getParent(),
                                path.getFileName());
            }
        });

        DataSetTestEndpoint endpoint = context.getEndpoint(
                String.format("dataset-test:file:%s?noop=true&split=true&timeout=1000&fileName=%s", path.getParent(),
                        path.getFileName()),
                DataSetTestEndpoint.class);
        endpoint.expectedMessageCount(4);

        producerTemplate.sendBody("direct:dataSetTestSplit", "Hello");
        producerTemplate.sendBody("direct:dataSetTestSplit", "World");
        producerTemplate.sendBody("direct:dataSetTestSplit", "Bye");
        producerTemplate.sendBody("direct:dataSetTestSplit", "World");

        try {
            endpoint.assertIsSatisfied(5000);
        } finally {
            context.removeRoute("dataSetTestSplit");
            endpoint.reset();
            Files.deleteIfExists(path);
        }
    }
}
