package com.github.tomitakussaari.mysqlcluscon;

import com.github.tomitakussaari.mysqlcluscon.galera.GaleraClusterConnectionChecker;
import com.github.tomitakussaari.mysqlcluscon.read_cluster.ReadClusterConnectionChecker;
import com.mysql.fabric.Server;

import java.lang.reflect.Proxy;
import java.net.URL;
import java.sql.*;
import java.util.*;
import java.util.logging.Logger;
import java.util.stream.Collectors;

import static com.github.tomitakussaari.mysqlcluscon.Params.DEFAULT_CONNECT_TIMEOUT_IN_MS;
import static com.github.tomitakussaari.mysqlcluscon.Params.MYSQL_CONNECT_TIMEOUT_PARAM;

public class MysclusconDriver implements Driver {

    private static final Logger LOGGER = Logger.getLogger(MysclusconDriver.class.getName());

    static {
        try {
            DriverManager.registerDriver(new MysclusconDriver());
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    static final String mysqlReadClusterConnectorName = "jdbc:myscluscon:mysql:read_cluster";
    static final String galeraClusterConnectorName = "jdbc:myscluscon:galera:cluster";

    private final ServerBlackList serverBlackList = new ServerBlackList();

    @Override
    public Connection connect(String jdbcUrl, Properties info) throws SQLException {
        if(acceptsURL(jdbcUrl)) {
            final Map<String, List<String>> queryParameters = URLHelpers.getQueryParameters(jdbcUrl);
            validateQueryParameters(queryParameters, jdbcUrl);
            final ConnectionStatus wantedConnectionStatus = getWantedConnectionStatus(queryParameters);
            final ConnectionChecker connectionChecker = chooseConnectionChecker(jdbcUrl, queryParameters);
            return createProxyConnection(connectionChecker, createActualConnection(jdbcUrl, connectionChecker, info, wantedConnectionStatus, queryParameters), wantedConnectionStatus);
        } else {
            return null;
        }
    }

    private void validateQueryParameters(Map<String, List<String>> queryParameters, String jdbcUrl) {
        if(!queryParameters.containsKey(MYSQL_CONNECT_TIMEOUT_PARAM)) {
            LOGGER.info(() -> "No connect timeout specified for URL: "+jdbcUrl+ " using default: "+DEFAULT_CONNECT_TIMEOUT_IN_MS);
            queryParameters.put(MYSQL_CONNECT_TIMEOUT_PARAM, Collections.singletonList(DEFAULT_CONNECT_TIMEOUT_IN_MS.toString()));
        }
    }

    @Override
    public boolean acceptsURL(String url) throws SQLException {
        return url.startsWith(mysqlReadClusterConnectorName) || url.startsWith(galeraClusterConnectorName);
    }

    @Override
    public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) throws SQLException {
        return new DriverPropertyInfo[0];
    }

    @Override
    public int getMajorVersion() {
        return 1;
    }

    @Override
    public int getMinorVersion() {
        return 0;
    }

    @Override
    public boolean jdbcCompliant() {
        return false;
    }

    @Override
    public Logger getParentLogger() {
        return LOGGER;
    }

    private Connection createActualConnection(String jdbcUrl, ConnectionChecker connectionChecker, Properties info, ConnectionStatus leastUsableConnection, Map<String, List<String>> queryParameters) throws SQLException {
        final List<String> hosts = serverBlackList.filterOutBlacklisted(URLHelpers.getHosts(jdbcUrl));
        return tryToOpenConnectionToValidHost(hosts, connectionChecker, info, jdbcUrl, leastUsableConnection, queryParameters)
                .orElseThrow(() -> new SQLException("Unable to open connection, no valid host found from hosts: " + hosts));
    }

    private Optional<Connection> tryToOpenConnectionToValidHost(List<String> hosts, ConnectionChecker connectionChecker,
                                                                Properties info, String jdbcUrl,
                                                                ConnectionStatus wantedConnectionStatus,
                                                                Map<String, List<String>> queryParameters) throws SQLException {
        LOGGER.fine(() -> "Trying to connect to hosts " + hosts + " from url " + jdbcUrl);
        Collections.shuffle(hosts);

        List<ConnectionAndStatus> connections = null;
        try {
            connections = hosts.stream()
                    .map(host -> tryConnectingToHost(host, jdbcUrl, info, queryParameters))
                    .filter(Optional::isPresent).map(Optional::get)
                    .map(conn -> new ConnectionAndStatus(conn, connectionChecker))
                    .collect(Collectors.toList());
            return findAndRemoveBestConnection(connections, wantedConnectionStatus);
        } finally {
            if(connections != null) {
                connections.forEach(ConnectionAndStatus::close);
            }
        }
    }

    private Optional<Connection> findAndRemoveBestConnection(List<ConnectionAndStatus> connections, ConnectionStatus leastUsableConnection) {
        Optional<ConnectionAndStatus> bestConnection = findBestConnection(connections, leastUsableConnection);
        bestConnection.ifPresent(connections::remove);
        return bestConnection.map(conn -> conn.connection);
    }

    Optional<ConnectionAndStatus> findBestConnection(List<ConnectionAndStatus> connections, ConnectionStatus wantedConnectionStatus) {
        return connections.stream()
            .filter(connectionAndStatus -> connectionAndStatus.getStatus().priority >= wantedConnectionStatus.priority)
            .sorted((left, right) -> right.getStatus().priority.compareTo(left.getStatus().priority))
            .findFirst();
    }

    private Optional<Connection> tryConnectingToHost(String host, String jdbcUrl, Properties info, Map<String, List<String>> queryParameters) {
        LOGGER.fine(() -> "Trying to connect to host " + host);
        final String connectUrl = URLHelpers.constructMysqlConnectUrl(host, jdbcUrl, queryParameters);
        try {
            LOGGER.fine(() -> "Connecting to " + connectUrl);
            return Optional.of(openRealConnection(info, connectUrl));
        } catch(Exception e) {
            serverBlackList.blackList(host);
            LOGGER.info(() -> "Error while opening connection " + connectUrl + " " + e.getMessage());
        }
        return Optional.empty();
    }


    protected Connection openRealConnection(Properties info, String connectUrl) throws SQLException {
        return DriverManager.getConnection(connectUrl, info);
    }


    private ConnectionChecker chooseConnectionChecker(String jdbcUrl, Map<String, List<String>> queryParameters) {
        String protocol = URLHelpers.getProtocol(jdbcUrl);
        LOGGER.fine(() -> "Parsed Protocol: " + protocol + " from url" + jdbcUrl);
        switch (protocol) {
            case mysqlReadClusterConnectorName: return new ReadClusterConnectionChecker(queryParameters);
            case galeraClusterConnectorName:    return new GaleraClusterConnectionChecker();
            default:                            throw new UnsupportedOperationException("Unsupported protocol: "+protocol);
        }
    }

    Connection createProxyConnection(final ConnectionChecker connectionChecker, Connection actualConnection, final ConnectionStatus wantedConnectionStatus) {
        return (Connection) Proxy.newProxyInstance(getClass().getClassLoader(), new Class[]{Connection.class}, (proxy, method, args) -> {
            switch(method.getName()) {
                case "isValid":
                    return connectionChecker.connectionStatus(actualConnection, (Integer) args[0]).priority >= wantedConnectionStatus.priority;
                default:
                    return method.invoke(actualConnection, args);
            }
        });
    }

    private ConnectionStatus getWantedConnectionStatus(Map<String, List<String>> queryParameters) {
        return ConnectionStatus.from(
                URLHelpers.getParameter(queryParameters, "connectionStatus", ConnectionStatus.STOPPED.toString())
        ).orElse(ConnectionStatus.STOPPED);
    }

}

@@ -409,16 +409,48 @@
 lectors;
+%0Aimport java.util.stream.Stream;
 %0A%0Aimport
@@ -4378,17 +4378,21 @@
        .
-m
+flatM
 ap(host 
@@ -4457,76 +4457,8 @@
 s))%0A
-                    .filter(Optional::isPresent).map(Optional::get)%0A
     
@@ -5635,32 +5635,30 @@
     private 
-Optional
+Stream
 %3CConnection%3E
@@ -6031,24 +6031,22 @@
  return 
-Optional
+Stream
 .of(open
@@ -6258,30 +6258,24 @@
 age());%0A    
-    %7D%0A
         retu
@@ -6277,24 +6277,22 @@
  return 
-Optional
+Stream
 .empty()
@@ -6289,24 +6289,34 @@
 am.empty();%0A
+        %7D%0A
     %7D%0A%0A%0A    

Replace map().filter().map() with single flatmap()