package eu.lundegaard.sdp.backend.controller;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import javax.servlet.http.HttpServletRequest;

/**
 * Kafka Connect API DEMO.
 * This class is used to populate the FE with sample Kafka Connect data.
 * @author Martin Kocisky
 */
@RestController
public class KafkaConnectController {

    @CrossOrigin
    @RequestMapping(value = "/connectors", method = RequestMethod.GET, produces = "application/json")
    public String getConnectors() {
        return "[\"my-jdbc-source\", \"my-hdfs-sink\"]";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors", method = RequestMethod.POST, consumes = "application/json", produces = "application/json")
    public String postConnectors(@RequestBody String data) {
        return "{\n" +
                "    \"name\": \"hdfs-sink-connector\",\n" +
                "    \"config\": {\n" +
                "        \"connector.class\": \"io.confluent.connect.hdfs.HdfsSinkConnector\",\n" +
                "        \"tasks.max\": \"10\",\n" +
                "        \"topics\": \"test-topic\",\n" +
                "        \"hdfs.url\": \"hdfs://fakehost:9000\",\n" +
                "        \"hadoop.conf.dir\": \"/opt/hadoop/conf\",\n" +
                "        \"hadoop.home\": \"/opt/hadoop\",\n" +
                "        \"flush.size\": \"100\",\n" +
                "        \"rotate.interval.ms\": \"1000\"\n" +
                "    },\n" +
                "    \"tasks\": [\n" +
                "        { \"connector\": \"hdfs-sink-connector\", \"task\": 1 },\n" +
                "        { \"connector\": \"hdfs-sink-connector\", \"task\": 2 },\n" +
                "        { \"connector\": \"hdfs-sink-connector\", \"task\": 3 }\n" +
                "    ]\n" +
                "}";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}", method = RequestMethod.GET, produces = "application/json")
    public String getConnectorWithName(@PathVariable String name, HttpServletRequest request) {
       return "{\n" +
               "    \"name\": \"hdfs-sink-connector\",\n" +
               "    \"config\": {\n" +
               "        \"connector.class\": \"io.confluent.connect.hdfs.HdfsSinkConnector\",\n" +
               "        \"tasks.max\": \"10\",\n" +
               "        \"topics\": \"test-topic\",\n" +
               "        \"hdfs.url\": \"hdfs://fakehost:9000\",\n" +
               "        \"hadoop.conf.dir\": \"/opt/hadoop/conf\",\n" +
               "        \"hadoop.home\": \"/opt/hadoop\",\n" +
               "        \"flush.size\": \"100\",\n" +
               "        \"rotate.interval.ms\": \"1000\"\n" +
               "    },\n" +
               "    \"tasks\": [\n" +
               "        { \"connector\": \"hdfs-sink-connector\", \"task\": 1 },\n" +
               "        { \"connector\": \"hdfs-sink-connector\", \"task\": 2 },\n" +
               "        { \"connector\": \"hdfs-sink-connector\", \"task\": 3 }\n" +
               "    ]\n" +
               "}";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}/config", method = RequestMethod.GET, produces = "application/json")
    public String getConnectorWithNameConfig(@PathVariable String name) {
        return "{\n" +
                "    \"connector.class\": \"io.confluent.connect.hdfs.HdfsSinkConnector\",\n" +
                "    \"tasks.max\": \"10\",\n" +
                "    \"topics\": \"test-topic\",\n" +
                "    \"hdfs.url\": \"hdfs://fakehost:9000\",\n" +
                "    \"hadoop.conf.dir\": \"/opt/hadoop/conf\",\n" +
                "    \"hadoop.home\": \"/opt/hadoop\",\n" +
                "    \"flush.size\": \"100\",\n" +
                "    \"rotate.interval.ms\": \"1000\"\n" +
                "}";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}/config", method = RequestMethod.PUT, consumes = "application/json", produces = "application/json")
    public String putConnectorWithNameConfig(@PathVariable String name) {
        return "{\n" +
                "    \"name\": \"hdfs-sink-connector\",\n" +
                "    \"config\": {\n" +
                "        \"connector.class\": \"io.confluent.connect.hdfs.HdfsSinkConnector\",\n" +
                "        \"tasks.max\": \"10\",\n" +
                "        \"topics\": \"test-topic\",\n" +
                "        \"hdfs.url\": \"hdfs://fakehost:9000\",\n" +
                "        \"hadoop.conf.dir\": \"/opt/hadoop/conf\",\n" +
                "        \"hadoop.home\": \"/opt/hadoop\",\n" +
                "        \"flush.size\": \"100\",\n" +
                "        \"rotate.interval.ms\": \"1000\"\n" +
                "    },\n" +
                "    \"tasks\": [\n" +
                "        { \"connector\": \"hdfs-sink-connector\", \"task\": 1 },\n" +
                "        { \"connector\": \"hdfs-sink-connector\", \"task\": 2 },\n" +
                "        { \"connector\": \"hdfs-sink-connector\", \"task\": 3 }\n" +
                "    ]\n" +
                "}";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}/status", method = RequestMethod.GET, produces = "application/json")
    public String getConnectorWithNameStatus(@PathVariable String name) {
        return "{\n" +
                "    \"name\": \"hdfs-sink-connector\",\n" +
                "    \"connector\": {\n" +
                "        \"state\": \"RUNNING\",\n" +
                "        \"worker_id\": \"fakehost:8083\"\n" +
                "    },\n" +
                "    \"tasks\":\n" +
                "    [\n" +
                "        {\n" +
                "            \"id\": 0,\n" +
                "            \"state\": \"RUNNING\",\n" +
                "            \"worker_id\": \"fakehost:8083\"\n" +
                "        },\n" +
                "        {\n" +
                "            \"id\": 1,\n" +
                "            \"state\": \"FAILED\",\n" +
                "            \"worker_id\": \"fakehost:8083\",\n" +
                "            \"trace\": \"org.apache.kafka.common.errors.RecordTooLargeException\\n\"\n" +
                "        }\n" +
                "    ]\n" +
                "}";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}/restart", method = RequestMethod.POST)
    public String postConnectorWithNameRestart(@PathVariable String name) {
        return "";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}/pause", method = RequestMethod.PUT)
    public String putConnectorWithNamePause(@PathVariable String name) {
        return "";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}/resume", method = RequestMethod.PUT)
    public String putConnectorWithNameResume(@PathVariable String name) {
        return "";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}", method = RequestMethod.DELETE)
    public String deleteConnectorWithName(@PathVariable String name) {
        return "";
    }


    /// TODO: TASKS
    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}/tasks", method = RequestMethod.GET)
    public String getConnectorWithNameTasks(@PathVariable String name) {
        return "[\n" +
                "    {\n" +
                "        \"task.class\": \"io.confluent.connect.hdfs.HdfsSinkTask\",\n" +
                "        \"topics\": \"test-topic\",\n" +
                "        \"hdfs.url\": \"hdfs://fakehost:9000\",\n" +
                "        \"hadoop.conf.dir\": \"/opt/hadoop/conf\",\n" +
                "        \"hadoop.home\": \"/opt/hadoop\",\n" +
                "        \"flush.size\": \"100\",\n" +
                "        \"rotate.interval.ms\": \"1000\"\n" +
                "    },\n" +
                "    {\n" +
                "        \"task.class\": \"io.confluent.connect.hdfs.HdfsSinkTask\",\n" +
                "        \"topics\": \"test-topic\",\n" +
                "        \"hdfs.url\": \"hdfs://fakehost:9000\",\n" +
                "        \"hadoop.conf.dir\": \"/opt/hadoop/conf\",\n" +
                "        \"hadoop.home\": \"/opt/hadoop\",\n" +
                "        \"flush.size\": \"100\",\n" +
                "        \"rotate.interval.ms\": \"1000\"\n" +
                "    }\n" +
                "]";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}/tasks/{taskId}/status", method = RequestMethod.GET)
    public String getConnectorWithName(@PathVariable String name, @PathVariable String taskId) {
        return "{\"state\":\"RUNNING\",\"id\":1,\"worker_id\":\"192.168.86.101:8083\"}\n";
    }

    @CrossOrigin
    @RequestMapping(value = "/connectors/{name}/tasks/{taskId}/restart", method = RequestMethod.POST)
    public String postConnectorWithNameTaskRestart(@PathVariable String name, @PathVariable String taskId) {
        return "";
    }

    /// TODO: PLUGINS

    @CrossOrigin
    @RequestMapping(value = "/connector-plugins", method = RequestMethod.GET)
    public String getConnectorPlugins() {
        return "[\n" +
                "    {\n" +
                "        \"class\": \"io.confluent.connect.hdfs.HdfsSinkConnector\"\n" +
                "    },\n" +
                "    {\n" +
                "        \"class\": \"io.confluent.connect.jdbc.JdbcSourceConnector\"\n" +
                "    }\n" +
                "]";
    }

    @CrossOrigin
    @RequestMapping(value = "/connector-plugins/{name}/config/validate", method = RequestMethod.PUT)
    public String putConnectorPluginsWithNameValidate(@PathVariable String name) {
        return "{\n" +
                "    \"name\": \"FileStreamSinkConnector\",\n" +
                "    \"error_count\": 1,\n" +
                "    \"groups\": [\n" +
                "        \"Common\"\n" +
                "    ],\n" +
                "    \"configs\": [\n" +
                "        {\n" +
                "            \"definition\": {\n" +
                "                \"name\": \"topics\",\n" +
                "                \"type\": \"LIST\",\n" +
                "                \"required\": false,\n" +
                "                \"default_value\": \"\",\n" +
                "                \"importance\": \"HIGH\",\n" +
                "                \"documentation\": \"\",\n" +
                "                \"group\": \"Common\",\n" +
                "                \"width\": \"LONG\",\n" +
                "                \"display_name\": \"Topics\",\n" +
                "                \"dependents\": [],\n" +
                "                \"order\": 4\n" +
                "        },\n" +
                "            \"value\": {\n" +
                "                \"name\": \"topics\",\n" +
                "                \"value\": \"test-topic\",\n" +
                "                \"recommended_values\": [],\n" +
                "                \"errors\": [],\n" +
                "                \"visible\": true\n" +
                "            }\n" +
                "        },\n" +
                "        {\n" +
                "            \"definition\": {\n" +
                "                \"name\": \"file\",\n" +
                "                \"type\": \"STRING\",\n" +
                "                \"required\": true,\n" +
                "                \"default_value\": \"\",\n" +
                "                \"importance\": \"HIGH\",\n" +
                "                \"documentation\": \"Destination filename.\",\n" +
                "                \"group\": null,\n" +
                "                \"width\": \"NONE\",\n" +
                "                \"display_name\": \"file\",\n" +
                "                \"dependents\": [],\n" +
                "                \"order\": -1\n" +
                "            },\n" +
                "            \"value\": {\n" +
                "                \"name\": \"file\",\n" +
                "                \"value\": null,\n" +
                "                \"recommended_values\": [],\n" +
                "                \"errors\": [\n" +
                "                    \"Missing required configuration \\\"file\\\" which has no default value.\"\n" +
                "                ],\n" +
                "                \"visible\": true\n" +
                "            }\n" +
                "        },\n" +
                "        {\n" +
                "            \"definition\": {\n" +
                "                \"name\": \"name\",\n" +
                "                \"type\": \"STRING\",\n" +
                "                \"required\": true,\n" +
                "                \"default_value\": \"\",\n" +
                "                \"importance\": \"HIGH\",\n" +
                "                \"documentation\": \"Globally unique name to use for this connector.\",\n" +
                "                \"group\": \"Common\",\n" +
                "                \"width\": \"MEDIUM\",\n" +
                "                \"display_name\": \"Connector name\",\n" +
                "                \"dependents\": [],\n" +
                "                \"order\": 1\n" +
                "            },\n" +
                "            \"value\": {\n" +
                "                \"name\": \"name\",\n" +
                "                \"value\": \"test\",\n" +
                "                \"recommended_values\": [],\n" +
                "                \"errors\": [],\n" +
                "                \"visible\": true\n" +
                "            }\n" +
                "        },\n" +
                "        {\n" +
                "            \"definition\": {\n" +
                "                \"name\": \"tasks.max\",\n" +
                "                \"type\": \"INT\",\n" +
                "                \"required\": false,\n" +
                "                \"default_value\": \"1\",\n" +
                "                \"importance\": \"HIGH\",\n" +
                "                \"documentation\": \"Maximum number of tasks to use for this connector.\",\n" +
                "                \"group\": \"Common\",\n" +
                "                \"width\": \"SHORT\",\n" +
                "                \"display_name\": \"Tasks max\",\n" +
                "                \"dependents\": [],\n" +
                "                \"order\": 3\n" +
                "        },\n" +
                "            \"value\": {\n" +
                "                \"name\": \"tasks.max\",\n" +
                "                \"value\": \"1\",\n" +
                "                \"recommended_values\": [],\n" +
                "                \"errors\": [],\n" +
                "                \"visible\": true\n" +
                "            }\n" +
                "        },\n" +
                "        {\n" +
                "            \"definition\": {\n" +
                "                \"name\": \"connector.class\",\n" +
                "                \"type\": \"STRING\",\n" +
                "                \"required\": true,\n" +
                "                \"default_value\": \"\",\n" +
                "                \"importance\": \"HIGH\",\n" +
                "                \"documentation\": \"Name or alias of the class for this connector. Must be a subclass of org.apache.kafka.connect.connector.Connector. If the connector is org.apache.kafka.connect.file.FileStreamSinkConnector, you can either specify this full name,  or use \\\"FileStreamSink\\\" or \\\"FileStreamSinkConnector\\\" to make the configuration a bit shorter\",\n" +
                "                \"group\": \"Common\",\n" +
                "                \"width\": \"LONG\",\n" +
                "                \"display_name\": \"Connector class\",\n" +
                "                \"dependents\": [],\n" +
                "                \"order\": 2\n" +
                "            },\n" +
                "            \"value\": {\n" +
                "                \"name\": \"connector.class\",\n" +
                "                \"value\": \"org.apache.kafka.connect.file.FileStreamSinkConnector\",\n" +
                "                \"recommended_values\": [],\n" +
                "                \"errors\": [],\n" +
                "                \"visible\": true\n" +
                "            }\n" +
                "        }\n" +
                "    ]\n" +
                "}";
    }

}
