/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.rakam.postgresql.analysis;

import com.facebook.presto.sql.RakamSqlFormatter;
import com.google.common.base.Throwables;
import org.rakam.analysis.AbstractFunnelQueryExecutor;
import org.rakam.analysis.metadata.Metastore;
import org.rakam.config.ProjectConfig;
import org.rakam.postgresql.report.PostgresqlQueryExecutor;
import org.rakam.util.ValidationUtil;

import javax.annotation.PostConstruct;
import javax.inject.Inject;

import java.sql.Connection;
import java.sql.SQLException;
import java.time.LocalDate;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

import static com.facebook.presto.sql.RakamExpressionFormatter.formatIdentifier;
import static java.lang.String.format;
import static org.rakam.util.ValidationUtil.checkTableColumn;

public class PostgresqlFunnelQueryExecutor
        extends AbstractFunnelQueryExecutor
{
    private final PostgresqlQueryExecutor executor;

    @Inject
    public PostgresqlFunnelQueryExecutor(ProjectConfig projectConfig, Metastore metastore, PostgresqlQueryExecutor executor)
    {
        super(projectConfig, metastore, executor);
        this.executor = executor;
    }

    @PostConstruct
    public void setup()
    {
        try (Connection conn = executor.getConnection()) {
            conn.createStatement().execute("CREATE OR REPLACE FUNCTION get_funnel_step(arr int[]) RETURNS integer AS $$\n" +
                    "DECLARE next_step integer := 1; step integer;\n" +
                    "        BEGIN \n" +
                    "                FOREACH step IN ARRAY arr\n" +
                    "       LOOP\n" +
                    "      IF step = next_step THEN\n" +
                    "         next_step = next_step + 1;\n" +
                    "      END IF;\n" +
                    "       END LOOP;\n" +
                    "    return next_step - 1;\n" +
                    "        END;\n" +
                    "$$ LANGUAGE plpgsql;");
        }
        catch (SQLException e) {
            throw Throwables.propagate(e);
        }
    }

    @Override
    public String getTemplate(List<FunnelStep> steps, Optional<String> dimension, Optional<FunnelWindow> window)
    {
        return "select %s get_funnel_step(steps) step, count(*) total from (\n" +
                "select %s array_agg(step order by " + checkTableColumn(projectConfig.getTimeColumn()) + ") as steps from (%s) t WHERE " + checkTableColumn(projectConfig.getTimeColumn()) + " between timestamp '%s' and timestamp '%s'\n" +
                "group by %s %s\n" +
                ") t group by 1 %s order by 1";
    }

    public String convertFunnel(String project, String connectorField, int idx, FunnelStep funnelStep, Optional<String> dimension, LocalDate startDate, LocalDate endDate)
    {
        String table = project + "." + ValidationUtil.checkCollection(funnelStep.getCollection());
        Optional<String> filterExp = funnelStep.getExpression().map(value -> RakamSqlFormatter.formatExpression(value,
                name -> name.getParts().stream().map(e -> formatIdentifier(e, '"')).collect(Collectors.joining(".")),
                name -> formatIdentifier("step" + idx, '"') + "." + name.getParts().stream()
                        .map(e -> formatIdentifier(e, '"')).collect(Collectors.joining(".")), '"'));

        String format = format("SELECT %s %s as _user, %d as step, %s from %s %s %s",
                dimension.map(ValidationUtil::checkTableColumn).map(v -> v + ",").orElse(""),
                format(connectorField, "step" + idx),
                idx + 1,
                checkTableColumn(projectConfig.getTimeColumn()),
                table,
                "step" + idx,
                filterExp.map(v -> "where " + v).orElse(""));
        return format;
    }
}


@@ -687,16 +687,71 @@
 wables;%0A
+import io.netty.handler.codec.http.HttpResponseStatus;%0A
 import o
@@ -934,32 +934,110 @@
 lQueryExecutor;%0A
+import org.rakam.report.QueryExecution;%0Aimport org.rakam.util.RakamException;%0A
 import org.rakam
@@ -1213,16 +1213,41 @@
 alDate;%0A
+import java.time.ZoneId;%0A
 import j
@@ -1648,16 +1648,79 @@
 xecutor;
+%0A    private final FastGenericFunnelQueryExecutor fastExecutor;
 %0A%0A    @I
@@ -1766,16 +1766,61 @@
 xecutor(
+FastGenericFunnelQueryExecutor fastExecutor, 
 ProjectC
@@ -1982,24 +1982,66 @@
 = executor;%0A
+        this.fastExecutor = fastExecutor;%0A
     %7D%0A%0A    @
@@ -2924,24 +2924,807 @@
    %7D%0A    %7D%0A%0A
+    @Override%0A    public QueryExecution query(String project, List%3CFunnelStep%3E steps, Optional%3CString%3E dimension, LocalDate startDate, LocalDate endDate, Optional%3CFunnelWindow%3E window, ZoneId zoneId, Optional%3CList%3CString%3E%3E connectors, Optional%3CBoolean%3E ordered)%0A    %7B%0A        if (!ordered.orElse(false)) %7B%0A            return fastExecutor.query(project, steps, dimension, startDate, endDate, window, zoneId, connectors, ordered);%0A        %7D%0A%0A        if (dimension.isPresent() && projectConfig.getUserColumn().equals(dimension.get())) %7B%0A            throw new RakamException(%22Dimension and connector field cannot be equal%22, HttpResponseStatus.BAD_REQUEST);%0A        %7D%0A%0A        return super.query(project, steps, dimension, startDate, endDate, window, zoneId, connectors, ordered);%0A    %7D%0A%0A
     @Overrid
@@ -5012,17 +5012,8 @@
 s %25s
- as _user
 , %25d

Fix postgresql funnel queyr bug