package yanagishima.servlet;

import me.geso.tinyorm.TinyORM;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import yanagishima.config.YanagishimaConfig;
import yanagishima.exception.HiveQueryErrorException;
import yanagishima.result.HiveQueryResult;
import yanagishima.row.Query;
import yanagishima.service.HiveService;
import yanagishima.util.AccessControlUtil;
import yanagishima.util.HttpRequestUtil;
import yanagishima.util.JsonUtil;
import yanagishima.util.MetadataUtil;

import javax.inject.Inject;
import javax.inject.Singleton;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.HashMap;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

import static javax.servlet.http.HttpServletResponse.SC_FORBIDDEN;
import static yanagishima.util.Constants.YANAGISHIMA_COMMENT;

@Singleton
public class HiveServlet extends HttpServlet {

    private static Logger LOGGER = LoggerFactory
            .getLogger(HiveServlet.class);

    private static final long serialVersionUID = 1L;

    @Inject
    private TinyORM db;

    private YanagishimaConfig yanagishimaConfig;

    private final HiveService hiveService;

    @Inject
    public HiveServlet(YanagishimaConfig yanagishimaConfig, HiveService hiveService) {
        this.yanagishimaConfig = yanagishimaConfig;
        this.hiveService = hiveService;
    }

    @Override
    protected void doPost(HttpServletRequest request,
                          HttpServletResponse response) throws ServletException, IOException {

        HashMap<String, Object> retVal = new HashMap<String, Object>();

        Optional<String> queryOptional = Optional.ofNullable(request.getParameter("query"));
        queryOptional.ifPresent(query -> {
            try {
                String userName = null;
                Optional<String> hiveUser = Optional.ofNullable(request.getParameter("user"));
                Optional<String> hivePassword = Optional.ofNullable(request.getParameter("password"));
                if(yanagishimaConfig.isUseAuditHttpHeaderName()) {
                    userName = request.getHeader(yanagishimaConfig.getAuditHttpHeaderName());
                } else {
                    if (hiveUser.isPresent() && hivePassword.isPresent()) {
                        userName = hiveUser.get();
                    }
                }
                if (yanagishimaConfig.isUserRequired() && userName == null) {
                    try {
                        response.sendError(SC_FORBIDDEN);
                        return;
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }

                String datasource = HttpRequestUtil.getParam(request, "datasource");
                if (yanagishimaConfig.isCheckDatasource()) {
                    if (!AccessControlUtil.validateDatasource(request, datasource)) {
                        try {
                            response.sendError(SC_FORBIDDEN);
                            return;
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }
                }
                if (userName != null) {
                    LOGGER.info(String.format("%s executed %s in %s", userName, query, datasource));
                }

                boolean storeFlag = Boolean.parseBoolean(Optional.ofNullable(request.getParameter("store")).orElse("false"));
                int limit = yanagishimaConfig.getSelectLimit();
                String engine = HttpRequestUtil.getParam(request, "engine");
                try {
                    HiveQueryResult hiveQueryResult = hiveService.doQuery(engine, datasource, query, userName, hiveUser, hivePassword, storeFlag, limit);
                    String queryid = hiveQueryResult.getQueryId();
                    retVal.put("queryid", queryid);
                    retVal.put("headers", hiveQueryResult.getColumns());
                    if(query.toLowerCase().indexOf("SHOW SCHEMAS") != -1) {
                        List<String> invisibleDatabases = yanagishimaConfig.getInvisibleDatabases(datasource);
                        retVal.put("results", hiveQueryResult.getRecords().stream().filter(list -> !invisibleDatabases.contains(list.get(0))).collect(Collectors.toList()));
                    } else {
                        retVal.put("results", hiveQueryResult.getRecords());
                    }
                    retVal.put("lineNumber", Integer.toString(hiveQueryResult.getLineNumber()));
                    retVal.put("rawDataSize", hiveQueryResult.getRawDataSize().toString());
                    Optional<String> warningMessageOptinal = Optional.ofNullable(hiveQueryResult.getWarningMessage());
                    warningMessageOptinal.ifPresent(warningMessage -> {
                        retVal.put("warn", warningMessage);
                    });
                    if(query.toLowerCase().startsWith("DESCRIBE")) {
                        if(yanagishimaConfig.getMetadataServiceUrl(datasource).isPresent()) {
                            String[] strings = query.toLowerCase().substring("DESCRIBE ".length()).split("\\.");
                            String schema = strings[0];
                            String table = strings[1].substring(1, strings[1].length() - 1);
                            MetadataUtil.setMetadata(yanagishimaConfig.getMetadataServiceUrl(datasource).get(), retVal, schema, table, hiveQueryResult.getRecords());
                        }
                    }
                } catch (HiveQueryErrorException e) {
                    LOGGER.error(e.getMessage());
                    retVal.put("queryid", e.getQueryId());
                    retVal.put("error", e.getCause().getMessage());
                }

            } catch (Throwable e) {
                LOGGER.error(e.getMessage(), e);
                retVal.put("error", e.getMessage());
            }
        });


        JsonUtil.writeJSON(response, retVal);

    }
}

@@ -4368,29 +4368,18 @@
 ery.
-toLowerCase().indexOf
+startsWith
 (%22SH
@@ -4394,14 +4394,8 @@
 AS%22)
- != -1
 ) %7B%0A
@@ -5299,30 +5299,16 @@
 f(query.
-toLowerCase().
 startsWi
@@ -5476,22 +5476,8 @@
 ery.
-toLowerCase().
 subs
@@ -5621,57 +5621,434 @@
 e = 
-strings%5B1%5D.substring(1, strings%5B1%5D.length() - 1);
+null;%0A                            if(engine.equals(%22hive%22)) %7B%0A                                table = strings%5B1%5D.substring(1, strings%5B1%5D.length() - 1);%0A                            %7D else if(engine.equals(%22spark%22)) %7B%0A                                table = strings%5B1%5D;%0A                            %7D else %7B%0A                                throw new IllegalArgumentException(engine + %22 is illegal%22);%0A                            %7D
 %0A   

remove back slash because it's not necessary in sparksql