package org.unipop.jdbc;

import com.google.common.collect.Sets;
import org.apache.commons.lang.NotImplementedException;
import org.jooq.SQLDialect;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.unipop.common.property.PropertySchema;
import org.unipop.common.property.PropertySchemaFactory;
import org.unipop.common.schema.referred.ReferredVertexSchema;
import org.unipop.jdbc.controller.schemas.RowEdgeSchema;
import org.unipop.jdbc.controller.schemas.RowVertexSchema;
import org.unipop.jdbc.simple.RowController;
import org.unipop.jdbc.utils.JdbcPredicatesTranslator;
import org.unipop.query.controller.SourceProvider;
import org.unipop.query.controller.UniQueryController;
import org.unipop.structure.UniGraph;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author Gur Ronen
 * @since 21/6/2016
 */
public class JdbcSourceProvider  implements SourceProvider {
    @Override
    public Set<UniQueryController> init(UniGraph graph, JSONObject configuration) throws Exception {
        Connection c = getConnection(configuration);
        SQLDialect dialect = SQLDialect.valueOf(configuration.getString("sqlDialect"));

        Set<RowVertexSchema> vertexSchemas = extractVertexSchemas(graph, configuration);
        Set<RowEdgeSchema> edgeSchemas = extractEdgeSchemas(graph, configuration);

        RowController controller = new RowController(graph, c, dialect, vertexSchemas, edgeSchemas, new JdbcPredicatesTranslator());
        return Sets.newHashSet(controller);
    }

    private Set<RowEdgeSchema> extractEdgeSchemas(UniGraph graph, JSONObject configuration) {
        List<JSONObject> edges = getConfigs(configuration, "edges");
        return edges.stream().map(edge -> {
            ReferredVertexSchema outVertexSchema = extractEdgeInnerVertexSchema(graph, edge, "outVertex");
            ReferredVertexSchema inVertexSchema = extractEdgeInnerVertexSchema(graph, edge, "inVertex");

            String table = edge.getString("table");
            List<PropertySchema> propertySchemas = PropertySchemaFactory.createPropertySchemas(edge);

            return new RowEdgeSchema(outVertexSchema, inVertexSchema, propertySchemas, graph, table);
        }).collect(Collectors.toSet());
    }

    private ReferredVertexSchema extractEdgeInnerVertexSchema(UniGraph graph, JSONObject edge, String key) {
        JSONObject vertexSchema = edge.getJSONObject(key);
        List<PropertySchema> propertySchemas = PropertySchemaFactory.createPropertySchemas(vertexSchema);
        return new ReferredVertexSchema(propertySchemas, graph);
    }

    @Override
    public void close() {
        throw new NotImplementedException();
    }

    private Set<RowVertexSchema> extractVertexSchemas(UniGraph graph, JSONObject configuration) {
        List<JSONObject> vertices = getConfigs(configuration, "vertices");
        return vertices.stream().map(vertex -> {
            String table = vertex.getString("table");
            List<PropertySchema> propertySchemas = PropertySchemaFactory.createPropertySchemas(vertex);

            return new RowVertexSchema(propertySchemas, graph, table);
        }).collect(Collectors.toSet());
    }

    private Connection getConnection(JSONObject configuration) throws SQLException {
        String url = configuration.getString("address");
        String user = configuration.getString("user");
        String password = configuration.getString("password");

        if (user == null && password == null) {
            return DriverManager.getConnection(url);
        }
        return DriverManager.getConnection(url, user, password);
    }

    private List<JSONObject> getConfigs(JSONObject configuration, String key) throws JSONException {
        List<JSONObject> configs = new ArrayList<>();
        JSONArray configsArray = configuration.optJSONArray(key);
        for(int i = 0; i < configsArray.length(); i++){
            JSONObject config = configsArray.getJSONObject(i);
            configs.add(config);
        }
        return configs;
    }
}

@@ -106,32 +106,59 @@
 entedException;%0A
+import org.jooq.Condition;%0A
 import org.jooq.
@@ -165,24 +165,24 @@
 SQLDialect;%0A
-
 import org.j
@@ -417,32 +417,84 @@
 edVertexSchema;%0A
+import org.unipop.common.util.PredicatesTranslator;%0A
 import org.unipo
@@ -1000,16 +1000,52 @@
 il.Set;%0A
+import java.util.function.Supplier;%0A
 import j
@@ -1122,16 +1122,16 @@
 016%0A */%0A
-
 public c
@@ -1183,16 +1183,398 @@
 vider %7B%0A
+    private final Supplier%3CPredicatesTranslator%3CIterable%3CCondition%3E%3E%3E predicatesTranslatorSupplier;%0A%0A    public JdbcSourceProvider() %7B%0A        this(JdbcPredicatesTranslator::new);%0A    %7D%0A%0A    public JdbcSourceProvider(Supplier%3CPredicatesTranslator%3CIterable%3CCondition%3E%3E%3E predicatesTranslatorSupplier) %7B%0A        this.predicatesTranslatorSupplier = predicatesTranslatorSupplier;%0A    %7D%0A%0A
     @Ove

add constructor to JdbcSourceProvider