package org.unipop.jdbc.controller.star;

import org.apache.tinkerpop.gremlin.process.traversal.Compare;
import org.apache.tinkerpop.gremlin.process.traversal.Contains;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.HasContainer;
import org.apache.tinkerpop.gremlin.process.traversal.util.MutableMetrics;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.T;
import org.jooq.*;
import org.jooq.impl.DSL;
import org.unipop.controller.Predicates;
import org.unipop.controller.VertexController;
import org.unipop.structure.BaseVertex;
import org.unipop.structure.UniGraph;

import java.sql.Connection;
import java.util.*;
import java.util.function.BiPredicate;

import static org.jooq.impl.DSL.field;
import static org.jooq.impl.DSL.table;

public class SqlTableController implements VertexController {

    private final DSLContext dslContext;
    private final UniGraph graph;
    private final Connection conn;
    private final String tableName;
    private final EdgeMapping[] mappings;
    private final VertexMapper vertexMapper;

    public SqlTableController(String tableName, UniGraph graph, Connection conn,  EdgeMapping... mappings) {
        this.graph = graph;
        this.conn = conn;
        this.tableName = tableName;
        this.mappings = mappings;
        dslContext = DSL.using(conn, SQLDialect.DEFAULT);
        vertexMapper = new VertexMapper();
    }

    public DSLContext getContext() {
        return dslContext;
    }

    @Override
    public Iterator<BaseVertex> vertices(Object[] ids) {
        return dslContext.select().from(tableName).where(field("id").in(ids)).fetch(vertexMapper).iterator();
    }

    @Override
    public Iterator<BaseVertex> vertices(Predicates predicates, MutableMetrics metrics) {
        SelectJoinStep<Record> select = dslContext.select().from(tableName);
        predicates.hasContainers.forEach(hasContainer -> select.where(createCondition(hasContainer)));
        //select.limit((int)predicates.limitLow, predicates.limitHigh < Long.MAX_VALUE ? (int)(predicates.limitHigh - predicates.limitLow) : Integer.MAX_VALUE);
        return select.fetch(vertexMapper).iterator();
    }

    private Condition createCondition(HasContainer hasContainer) {
        String key = hasContainer.getKey();
        Object value = hasContainer.getValue();
        BiPredicate<?, ?> predicate = hasContainer.getBiPredicate();

        if(key.equals("~id"))
            return field(T.id.toString()).in(value.getClass().isArray() ? (Object[])value : new Object[]{value});
        Field<Object> field = field(key);
        if (predicate instanceof Compare) {
            String predicateString = predicate.toString();
            switch (predicateString) {
                case ("eq"):
                    return field.eq(value);
                case ("neq"):
                    return field.notEqual(value);
                case ("gt"):
                    return field.greaterThan(value);
                case ("gte"):
                    return field.greaterOrEqual(value);
                case ("lt"):
                    return field.lessThan(value);
                case ("lte"):
                    return field.lessOrEqual(value);
                case("inside"):
                    List items =(List) value;
                    Object firstItem = items.get(0);
                    Object secondItem = items.get(1);
                    return field.between(firstItem, secondItem);
                default:
                    throw new IllegalArgumentException("predicate not supported in has step: " + predicate.toString());
            }
        } else if (predicate instanceof Contains) {
            if (predicate == Contains.without) return field.isNull();
            else if (predicate == Contains.within){
                if(value == null) return field.isNotNull();
               // else
            }
        }
        throw new IllegalArgumentException("predicate not supported by unipop: " + predicate.toString());
    }

    @Override
    public BaseVertex fromEdge(Direction direction, Object vertexId, String vertexLabel) {
        return dslContext.select().from(tableName).where(field("id").eq(vertexId)).fetchOne(vertexMapper);
    }

    @Override
    public BaseVertex addVertex(Object id, String label, Object[] properties) {
        Collection<Field<Object>> fields = new ArrayList<>();
        List<Object> values = new ArrayList<>();

        for(int i = 0; i < properties.length; i = i+2){
            if(properties[i].toString().equals("label")) continue;
            fields.add(field(properties[i].toString()));
            values.add(properties[i+1]);
        }

        dslContext.insertInto(table(tableName), fields).values(values).execute();
        return new SqlVertex(id, label, properties, this, graph);
    }

    private SqlTableController self = this;
    private class VertexMapper implements RecordMapper<Record, BaseVertex> {

        @Override
        public BaseVertex map(Record record) {
            Map<String, Object> stringObjectMap = record.intoMap();
            SqlVertex vertex = new SqlVertex(stringObjectMap.get("ID"), tableName.toLowerCase(), null, self, graph);
            stringObjectMap.forEach((key, value) -> {
                if(value != null)
                    vertex.addPropertyLocal(key, value);
            });
            return vertex;
        }
    }
}

@@ -35,16 +35,72 @@
 .star;%0A%0A
+import org.apache.commons.collections4.CollectionUtils;%0A
 import o
@@ -734,17 +734,69 @@
 va.util.
-*
+Iterator;%0Aimport java.util.List;%0Aimport java.util.Map
 ;%0Aimport
@@ -2536,24 +2536,99 @@
 edicate();%0A%0A
+        if(key.equals(T.label.getAccessor())) return DSL.trueCondition();%0A%0A
         if(k
@@ -4518,25 +4518,36 @@
 g label,
+ Map%3CString,
  Object
-%5B%5D
+%3E
  propert
@@ -4565,292 +4565,151 @@
     
-Collection%3CField%3CObject%3E%3E fields = new ArrayList%3C%3E();%0A        List%3CObject%3E values = new ArrayList%3C%3E();%0A%0A        for(int i = 0; i %3C properties.length; i = i+2)%7B%0A            if(properties%5Bi%5D.toString().equals(%22label%22)) c
+properties.putIfAbsent(%22id%22, id);%0A%0A        dslContext.insertInto(table(tableName), Collecti
 on
+U
 ti
-nue;%0A            fields.add(field(properties%5Bi%5D.toString()));%0A
+ls.collect(properties.keySet(), DSL::field))%0A  
     
@@ -4720,18 +4720,17 @@
     
+  .
 values
-.add
 (pro
@@ -4740,90 +4740,16 @@
 ties
-%5Bi+1%5D);%0A        %7D%0A%0A        dslContext.insertInto(table(tableName), fields)
 .values(
 valu
@@ -4744,22 +4744,17 @@
 .values(
-values
+)
 ).execut
@@ -4754,24 +4754,25 @@
 .execute();%0A
+%0A
         retu
@@ -5103,26 +5103,14 @@
     
-SqlVertex vertex =
+return
  new
@@ -5176,214 +5176,37 @@
 (), 
-null, self, graph);%0A            stringObjectMap.forEach((key, value) -%3E %7B%0A                if(value != null)%0A                    vertex.addPropertyLocal(key, value);%0A            %7D);%0A            return vertex
+stringObjectMap, self, graph)
 ;%0A  

fix id & label fields problems.