"""Kafka connect heplers."""
import json
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple

import requests

from dbdeploy.base import constants
from dbdeploy.util.aws.helpers import get_secret_name


def create_connector(
        uri: str,
        connector_name: str,
        connector_config: Dict[str, str]) -> Tuple[bool, Optional[str]]:
    """Create kafka connect connector using rest interface.

    Args:
        uri The URI of the Kafka cluster
        connector_name: The name of the connector to create
        connector_config: The configuration of the connector being created

    Returns:
        boolean value and an error message in case there is an error.
    """
    data = json.dumps(connector_config)
    headers = {"Content-Type": "application/json"}
    connector_route = constants.CONNECTOR_ROUTE_TEMPLATE.format(connector_name)
    url = f'{uri}{connector_route}/config'

    response = requests.put(url=url, headers=headers, data=data)
    if response.status_code not in range(200, 300):
        msg = ('Cannot create connector using provided configuration. '
               f'Status code: {response.status_code}.')
        return False, msg
    return True, None


def delete_connector(
        uri: str,
        connector_name: str) -> Tuple[bool, Optional[str]]:
    """Delete kafka connect connector using rest interface.

    Args:
        uri The URI of the Kafka cluster
        connector_name: The name of the connector to create

    Returns:
        boolean value and an error message in case there is an error.
    """
    connector_route = constants.CONNECTOR_ROUTE_TEMPLATE.format(connector_name)
    url = f'{uri}{connector_route}'

    response = requests.delete(url=url)
    if response.status_code not in range(200, 300):
        msg = ('Cannot delete connector. '
               f'Status code: {response.status_code}.')
        return False, msg
    return True, None


def get_connector_list(
        uri: str) -> Tuple[List[str], Optional[str]]:
    """Get running connectors count using rest interface.

    Args:
        uri The URI of the Kafka cluster

    Returns:
        connector count and error message in case there is an error.
    """
    connectors_route = '/connectors/'
    url = f'{uri}{connectors_route}'

    response = requests.get(url=url)
    if not response.ok:
        msg = ('Cannot get connector list. Assuming 0 connectors.'
               f'Status code: {response.status_code}.')
        return [], msg
    return response.json(), None


def get_neo4j_sink_connector_config(
            cypher_query: str,
            dlq_topic_name: str,
            environment: str,
            kafka_bootstrap_brokers: str,
            neo4j_server_uri: str,
            topic: str,
            batch_size: int = 2500,
            tasks_max: int = 1,
            max_poll_interval: Optional[int] = None,
            aws_region: str = 'us-east-1',
            kafka_security_protocol: str = 'SSL',
            service_name: str = 'kafka-db-deploy',
            secret_key: str = 'NEO4J_CREDENTIALS'
        ) -> Dict[str, str]:
    """Generate connector configuration.

    Args:
        cypher_query: The cypher query to use in the topic-cypher mapping
        dlq_topic_name: The DeadLetterQueue topic name
        environment: The name of the environment
        kafka_bootstrap_brokers: Kafka cluster bootstrap brokers
        neo4j_server_uri: Neo4j cluster connection uri
        topic: The name of the topic to consume from
        batch_size: The size of the batch connector's neo4j executor
            is going to UNWIND (default: 2500)
        tasks_max: The number of connector tasks (processes) (default: 1)
        aws_region: The AWS region (default: 'us-east-1')
        kafka_security_protocol: Kafka security protocol to use (default: SSL)
        service_name: Kafka-connect cluster name which is the
            same as fargate service name (default: kafka-db-deploy)
        secret_key: The secret key for connector secrets.

    Returns:
        A dict with kafka-connect task group configuration.

    """
    secret_name = get_secret_name(
        environment,
        service_name,
        secret_key)
    batch_timeout = max_poll_interval
    if not max_poll_interval:
        # if that's a normal run set default values
        max_poll_interval = 300000  # ms
        batch_timeout = 300000  # ms
    return {
        'tasks.max': str(tasks_max),

        # Topic and topic mapping configuration
        'topics': topic,
        f'neo4j.cypher.topic.{topic}': cypher_query,

        # DLQ configuration
        'errors.deadletterqueue.topic.name': dlq_topic_name,
        'kafka.bootstrap.servers': kafka_bootstrap_brokers,
        'kafka.security.protocol': kafka_security_protocol,

        # Neo4j executor configuration
        'neo4j.uri': neo4j_server_uri,
        'neo4j.batch-size': str(batch_size),
        'neo4j.batch-timeout': f'{batch_timeout}ms',

        # This places an upper bound on the amount of time that the consumer
        # can be idle before fetching more records. If poll() is not called
        # before expiration of this timeout, then the consumer is considered
        # failed and the group will rebalance in order to reassign the
        # partitions to another member.
        # ref: https://kafka.apache.org/documentation/#consumerconfigs
        'consumer.override.max.poll.interval.ms': str(max_poll_interval),

        # Secrets provider configuration
        'config.providers': 'aws',
        'config.providers.aws.class': 'io.lenses.connect.secrets.providers.AWSSecretProvider',  # noqa: E501
        'config.providers.aws.param.aws.auth.method': 'default',
        'config.providers.aws.param.aws.access.key': 'dummy-client-key',
        'config.providers.aws.param.aws.secret.key': 'dummy-secret-key',
        'config.providers.aws.param.aws.region': aws_region,
        'neo4j.authentication.basic.username': '${aws:'f'{secret_name}'':username}',  # noqa: E501
        'neo4j.authentication.basic.password': '${aws:'f'{secret_name}'':password}',  # noqa: E501

        # Non-configurable static params
        'connector.class': 'org.neo4j.connectors.kafka.sink.Neo4jConnector',
        'neo4j.max-retry-time': '120s',  # total time spent retrying transient errors
        'key.converter': 'org.apache.kafka.connect.storage.StringConverter',
        'value.converter': 'org.apache.kafka.connect.json.JsonConverter',
        'key.converter.schemas.enable': 'false',
        'value.converter.schemas.enable': 'false',
        'errors.retry.timeout': '-1',
        'neo4j.encryption.enabled': 'true',
        'errors.retry.delay.max.ms': '1000',
        'errors.tolerance': 'all',
        'errors.deadletterqueue.context.headers.enable': 'true',
    }


def get_jdbc_sink_connector_config(
            environment: str,
            topic: str,
            dlq_topic_name: str,
            table_name: str,
            primary_keys: str,
            insert_mode: str | None = 'update',
            batch_size: int = 2500,
            tasks_max: int = 1,
            aws_region: str = 'us-east-1',
            service_name: str = 'kafka-db-deploy',
            secret_key: str = 'MYSQL_CREDENTIALS'
        ) -> Dict[str, str]:
    """Generate connector configuration.

    Args:
        environment: The name of the environment
        topic: The name of the topic to consume from
        dlq_topic_name: The DeadLetterQueue topic name
        table_name: The table name in the destination Mysql
        primary_keys: A string with comma-separated list of primary keys
        insert_mode: JDBC insert mode (insert, upsert, update, delete)
        batch_size: A batch size to be processed by connector
        tasks_max: The number of connector tasks (processes) (default: 1)
        aws_region: The AWS region (default: 'us-east-1')
        service_name: The name of the service
        secret_key: The secret key for connector secrets.

    Returns:
        A dict with kafka-connect task group configuration.

    """
    # Validate insert_mode
    from dbdeploy.base import constants
    if insert_mode not in constants.JDBC_INSERT_MODES:
        raise ValueError(f"Invalid insert_mode '{insert_mode}'. Must be one of: {constants.JDBC_INSERT_MODES}")

    secret_name = get_secret_name(
        environment,
        service_name,
        secret_key)

    return {
        'tasks.max': str(tasks_max),

        # Secrets provider configuration
        'config.providers': 'aws',
        'config.providers.aws.class': 'io.lenses.connect.secrets.providers.AWSSecretProvider',  # noqa: E501
        'config.providers.aws.param.aws.auth.method': 'default',
        'config.providers.aws.param.aws.access.key': 'dummy-client-key',
        'config.providers.aws.param.aws.secret.key': 'dummy-secret-key',
        'config.providers.aws.param.aws.region': aws_region,
        'connection.user': '${aws:'f'{secret_name}'':username}',
        'connection.password': '${aws:'f'{secret_name}'':password}',
        'connection.url': 'jdbc:mysql://${aws:'f'{secret_name}'':uri}',  # noqa: E501

        # Topic configuration
        'topics': topic,

        # DLQ configuration
        'errors.deadletterqueue.topic.name': dlq_topic_name,

        'connector.class': 'io.confluent.connect.jdbc.JdbcSinkConnector',
        'key.converter': 'org.apache.kafka.connect.storage.StringConverter',
        'value.converter': 'org.apache.kafka.connect.json.JsonConverter',
        'key.converter.schemas.enable': 'false',
        'value.converter.schemas.enable': 'true',

        'transforms': 'ValueToKey,RegexRouter',
        'transforms.ValueToKey.type': 'org.apache.kafka.connect.transforms.ValueToKey',  # noqa: E501
        'transforms.ValueToKey.fields': f'{primary_keys}',
        'transforms.RegexRouter.type': 'org.apache.kafka.connect.transforms.RegexRouter',  # noqa: E501
        'transforms.RegexRouter.regex': '.*',
        'transforms.RegexRouter.replacement': f'{table_name}',

        'pk.mode': 'record_key',
        'pk.fields': '',

        'insert.mode': insert_mode,
        'auto.create': 'false',
        'auto.evolve': 'false',
        'dialect.name': 'MySqlDatabaseDialect',
        'table.name.format': '${topic}',
        'delete.enabled': 'false',

        'batch.size': f'{batch_size}',
        'errors.tolerance': 'all',
    }