package org.rakam.postgresql.plugin.user;

import com.facebook.presto.sql.tree.QualifiedName;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import io.netty.handler.codec.http.HttpResponseStatus;
import org.rakam.analysis.metadata.Metastore;
import org.rakam.plugin.EventStore;
import org.rakam.plugin.user.AbstractUserService;
import org.rakam.plugin.user.UserPluginConfig;
import org.rakam.plugin.user.UserStorage;
import org.rakam.postgresql.report.PostgresqlQueryExecutor;
import org.rakam.report.QueryExecution;
import org.rakam.report.QueryResult;
import org.rakam.util.JsonHelper;
import org.rakam.util.RakamException;

import javax.inject.Inject;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.lang.String.format;
import static org.rakam.util.ValidationUtil.checkProject;

public class PostgresqlUserService extends AbstractUserService {
    private final Metastore metastore;
    private final PostgresqlQueryExecutor executor;
    private final PostgresqlUserStorage storage;

    @Inject
    public PostgresqlUserService(PostgresqlUserStorage storage, UserPluginConfig config, EventStore eventStore, Metastore metastore, PostgresqlQueryExecutor executor) {
        super(storage, config, eventStore);
        this.storage = storage;
        this.metastore = metastore;
        this.executor = executor;
    }

    @Override
    public CompletableFuture<List<CollectionEvent>> getEvents(String project, String user, int limit, Instant beforeThisTime) {
        checkProject(project);
        checkNotNull(user);
        checkArgument(limit <= 1000, "Maximum 1000 events can be fetched at once.");
        String sqlQuery = metastore.getCollections(project).entrySet().stream()
                .filter(entry -> entry.getValue().stream().anyMatch(field -> field.getName().equals("_user")))
                .filter(entry -> entry.getValue().stream().anyMatch(field -> field.getName().equals("_time")))
                .map(entry ->
                        format("select '%s' as collection, row_to_json(coll) json, _time from \"%s\".\"%s\" coll where _user = '%s' %s",
                                entry.getKey(), project, entry.getKey(), user, beforeThisTime == null ? "" : String.format("and _time < timestamp '%s'", beforeThisTime.toString())))
                .collect(Collectors.joining(" union all "));

        if (sqlQuery.isEmpty()) {
            return CompletableFuture.completedFuture(ImmutableList.<CollectionEvent>of());
        }

        CompletableFuture<QueryResult> queryResult = executor.executeRawQuery(format("select collection, json from (%s) data order by _time desc limit %d", sqlQuery, limit)).getResult();
        return queryResult.thenApply(result -> {
            if (result.isFailed()) {
                throw new RakamException(result.getError().toString(), HttpResponseStatus.INTERNAL_SERVER_ERROR);
            }

            List<CollectionEvent> events = new ArrayList(result.getResult().size());
            for (List<Object> objects : result.getResult()) {
                events.add(new CollectionEvent((String) objects.get(0), JsonHelper.read(objects.get(1).toString(), Map.class)));
            }

            return events;
        });
    }

    @Override
    public void merge(String project, Object user, Object anonymousId, Instant createdAt, Instant mergedAt) {
        try (Connection connection = executor.getConnection()) {
            for (String collectionName : metastore.getCollectionNames(project)) {
                PreparedStatement ps = connection.prepareStatement(String.format("UPDATE %s SET _user = ? WHERE _user = ? AND _time BETWEEN ? and ?",
                        executor.formatTableReference(project, QualifiedName.of(collectionName))));
                storage.setUserId(project, ps, user, 1);
                storage.setUserId(project, ps, anonymousId, 2);
                ps.setTimestamp(3, Timestamp.from(createdAt));
                ps.setTimestamp(4, Timestamp.from(createdAt));
                ps.executeUpdate();
            }
        } catch (SQLException e) {
            throw Throwables.propagate(e);
        }
    }

    @Override
    public QueryExecution precalculate(String project, PreCalculateQuery query) {
        // no-op
        return QueryExecution.completedQueryExecution(null, QueryResult.empty());
    }
}

@@ -274,24 +274,65 @@
 .Metastore;%0A
+import org.rakam.collection.SchemaField;%0A
 import org.r
@@ -732,16 +732,17 @@
 Inject;%0A
+%0A
 import j
@@ -1299,16 +1299,24 @@
 rService
+%0A       
  extends
@@ -1335,17 +1335,17 @@
 rService
- 
+%0A
 %7B%0A    pr
@@ -1657,16 +1657,20 @@
 xecutor)
+%0A   
  %7B%0A     
@@ -1956,16 +1956,20 @@
 hisTime)
+%0A   
  %7B%0A     
@@ -2685,23 +2685,16 @@
  ? %22%22 : 
-String.
 format(%22
@@ -3809,16 +3809,20 @@
 ergedAt)
+%0A   
  %7B%0A     
@@ -3828,142 +3828,299 @@
     
-try (Connection connection = execu
+for (Map.Entry%3CString, List%3CSchemaField%3E%3E entry : metas
 tor
+e
 .getCo
-nn
+ll
 ection
-()) %7B%0A            for (String collectionName : metas
+s(project).entrySet()) %7B%0A            if (!entry.getValue().stream().anyMatch(e -%3E e.getName().equals(%22_user%22))) %7B%0A                continue;%0A            %7D%0A            try (Connection connection = execu
 tor
-e
 .getCo
-ll
+nn
 ection
-Names(project
+(
 )) %7B
@@ -4187,23 +4187,16 @@
 atement(
-String.
 format(%22
@@ -4347,22 +4347,22 @@
 .of(
-collectionName
+entry.getKey()
 ))))
@@ -4668,17 +4668,19 @@
         
-%7D
+   
  catch (
@@ -4693,24 +4693,28 @@
 eption e) %7B%0A
+    
             
@@ -4740,24 +4740,38 @@
 opagate(e);%0A
+            %7D%0A
         %7D%0A  
@@ -4868,16 +4868,20 @@
 y query)
+%0A   
  %7B%0A     

check _user column before merging users