/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.apache.samza.sql.client.impl;

import java.io.File;
import java.io.PrintWriter;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.stream.Collectors;
import org.apache.samza.sql.client.cli.CliCommand;
import org.apache.samza.sql.client.cli.CliConstants;
import org.apache.samza.sql.client.cli.CliEnvironment;
import org.apache.samza.sql.client.cli.CliShell;
import org.apache.samza.sql.client.cli.CliView;
import org.apache.samza.sql.client.cli.QueryResultLogView;
import org.apache.samza.sql.client.exceptions.CommandHandlerException;
import org.apache.samza.sql.client.exceptions.ExecutorException;
import org.apache.samza.sql.client.interfaces.CommandHandler;
import org.apache.samza.sql.client.interfaces.ExecutionContext;
import org.apache.samza.sql.client.interfaces.ExecutionStatus;
import org.apache.samza.sql.client.interfaces.NonQueryResult;
import org.apache.samza.sql.client.interfaces.QueryResult;
import org.apache.samza.sql.client.interfaces.SqlExecutor;
import org.apache.samza.sql.client.interfaces.SqlFunction;
import org.apache.samza.sql.client.util.CliUtil;
import org.apache.samza.sql.schema.SamzaSqlFieldType;
import org.apache.samza.sql.schema.SqlFieldSchema;
import org.apache.samza.sql.schema.SqlSchema;
import org.jline.terminal.Terminal;
import org.jline.utils.AttributedStringBuilder;
import org.jline.utils.AttributedStyle;
import org.jline.utils.InfoCmp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * Default {@link CommandHandler} which handles basic commands of type {@link CliCommandType}
 */
public class CliCommandHandler implements CommandHandler {
  private static final Logger LOG = LoggerFactory.getLogger(CliCommandHandler.class);
  private CliShell shell;
  private PrintWriter writer;
  private Terminal terminal;
  private ExecutionContext exeContext;
  private SqlExecutor executor;
  private CliEnvironment env;
  private Map<Integer, String> executions = new TreeMap<>();

  /**
   * sets up the member variables
   * @param shell the {@link CliShell} which uses this CommandHandler
   * @param env the Shell's {@link CliEnvironment}
   * @param terminal the {@link Terminal} to print output and messages
   * @param exeContext the {@link ExecutionContext}
   */
  public void init(CliShell shell, CliEnvironment env, Terminal terminal, ExecutionContext exeContext) {
    this.env = env;
    executor = env.getExecutor();
    this.terminal = terminal;
    this.writer = terminal.writer();
    this.exeContext = exeContext;
    this.shell = shell;
  }

  /**
   * Attempts to parse the given input string line into a {@link CliCommand} of this
   * handler's {@link org.apache.samza.sql.client.interfaces.CommandType}
   * @param line input line string
   * @return {@link CliCommand} on success, null otherwise
   */
  public CliCommand parseLine(String line) {
    line = CliUtil.trimCommand(line);
    if (CliUtil.isNullOrEmpty(line))
      return null;

    String upperCaseLine = line.toUpperCase();
    for (CliCommandType cmdType : CliCommandType.values()) {
      String cmdText = cmdType.getCommandName();
      if (upperCaseLine.startsWith(cmdText)) {
        if (upperCaseLine.length() == cmdText.length())
          return new CliCommand(cmdType);
        else if (upperCaseLine.charAt(cmdText.length()) <= CliConstants.SPACE) {
          String parameter = line.substring(cmdText.length()).trim();
          if (!parameter.isEmpty())
            return new CliCommand(cmdType, parameter);
        }
      }
    }
    return new CliCommand(CliCommandType.INVALID_COMMAND);
  }

  /**
   * Prints to terminal the help message of the commands this handler handles
   */
  public void printHelpMessage() {
    AttributedStringBuilder builder = new AttributedStringBuilder();
    builder.append("The following commands are supported by ")
        .append(CliConstants.APP_NAME)
        .append("by handler ")
        .append(this.getClass().getName())
        .append(" at the moment.\n\n");

    for (CliCommandType cmdType : CliCommandType.values()) {
      if (cmdType == CliCommandType.INVALID_COMMAND)
        continue;

      String cmdText = cmdType.getCommandName();
      String cmdDescription = cmdType.getDescription();

      builder.style(AttributedStyle.DEFAULT.bold())
          .append(cmdText)
          .append("\t\t")
          .style(AttributedStyle.DEFAULT)
          .append(cmdDescription)
          .append("\n");
    }
    writer.println(builder.toAnsi());
  }

  /**
   * Handles the given command
   * @param command input {@link CliCommand} to handle
   * @return false if command is to quit, or fatal error happened that Shell should not continue running. True o.w.
   * @throws CommandHandlerException if unrecoverable error happened while handling the input {@link CliCommand}
   */
  public boolean handleCommand(CliCommand command) throws CommandHandlerException {
    boolean keepRunning = true;

    if (!command.getCommandType().argsAreOptional() && CliUtil.isNullOrEmpty(command.getParameters())) {
      CliUtil.printCommandUsage(command, writer);
      return true;
    }

    try {
      switch ((CliCommandType) command.getCommandType()) {
        case CLEAR:
          commandClear();
          break;

        case DESCRIBE:
          commandDescribe(command);
          break;

        case EXECUTE:
          commandExecuteFile(command);
          break;

        case EXIT:
        case QUIT:
          keepRunning = false;
          break;

        case HELP:
          commandHelp(command);
          break;

        case INSERT_INTO:
          commandInsertInto(command);
          break;

        case LS:
          commandLs(command);
          break;

        case RM:
          commandRm(command);
          break;

        case SELECT:
          commandSelect(command);
          break;

        case SET:
          commandSet(command);
          break;

        case SHOW_FUNCTIONS:
          commandShowFunctions(command);
          break;

        case SHOW_TABLES:
          commandShowTables(command);
          break;

        case STOP:
          commandStop(command);
          break;

        case VERSION:
          commandVersion();
          break;

        case INVALID_COMMAND:
          printHelpMessage();
          break;

        default:
          writer.println("UNDER DEVELOPEMENT. Command:" + command.getCommandType());
          writer.println("Parameters:" + (CliUtil.isNullOrEmpty(command.getParameters()) ? "NULL" : command.getParameters()));
          writer.flush();
      }
    } catch (Exception e) {
      throw new CommandHandlerException(e);
    }
    return keepRunning;
  }


  private void commandVersion() {
    printVersion();
  }

  private void printVersion() {
    String version = String.format("Shell version %s, Executor is %s, version %s",
        this.getClass().getPackage().getImplementationVersion(),
        executor.getClass().getName(),
        executor.getVersion());
    writer.println(version);
  }

  private void commandClear() {
    clearScreen();
  }

  private void commandDescribe(CliCommand command) throws ExecutorException {
    String parameters = command.getParameters();
    SqlSchema schema = executor.getTableSchema(exeContext, parameters);
    List<String> lines = formatSchema4Display(schema);
    for (String line : lines) {
      writer.println(line);
    }
    writer.flush();
  }

  private void commandSet(CliCommand command) throws Exception {
    String param = command.getParameters();
    if (CliUtil.isNullOrEmpty(param)) {
      env.printAll(writer);
      writer.flush();
      return;
    }
    String[] params = null;
    boolean syntaxValid = param.split(" ").length == 1;
    if (syntaxValid) {
      params = param.split("=");
      if (params.length == 1) {
        String value = env.getEnvironmentVariable(param);
        if (!CliUtil.isNullOrEmpty(value)) {
          env.printVariable(writer, param, value);
        }
        return;
      } else {
        syntaxValid = params.length == 2;
      }
    }
    if (!syntaxValid) {
      writer.println(command.getCommandType().getUsage());
      writer.flush();
      return;
    }

    String name = params[0].trim().toLowerCase();
    String value = params[1].trim();
    int ret = env.setEnvironmentVariable(name, value);
    if (ret == 0) {
      writer.print(name);
      writer.print(" set to ");
      writer.println(value);
      if (name.equals(CliConstants.CONFIG_EXECUTOR)) {
        executor.stop(exeContext);
        executor = env.getExecutor();
        executor.start(exeContext);
      }
    } else if (ret == -1) {
      writer.print("Unknow variable: ");
      writer.println(name);
    } else if (ret == -2) {
      writer.print("Invalid value: ");
      writer.print(value);
      String[] vals = env.getPossibleValues(name);
      if (vals != null && vals.length != 0) {
        writer.print(" Possible values:");
        for (String s : vals) {
          writer.print(CliConstants.SPACE);
          writer.print(s);
        }
        writer.println();
      }
    }

    writer.flush();
  }

  private void commandExecuteFile(CliCommand command) throws ExecutorException {
    String fullCmdStr = command.getFullCommand();
    String parameters = command.getParameters();
    URI uri = null;
    boolean valid = false;
    File file = null;
    try {
      uri = new URI(parameters);
      file = new File(uri.getPath());
      valid = file.exists() && !file.isDirectory();
    } catch (URISyntaxException e) {
    }
    if (!valid) {
      writer.println("Invalid URI.");
      writer.flush();
      return;
    }

    NonQueryResult nonQueryResult = executor.executeNonQuery(exeContext, file);
    executions.put(nonQueryResult.getExecutionId(), fullCmdStr);
    List<String> submittedStmts = nonQueryResult.getSubmittedStmts();
    List<String> nonsubmittedStmts = nonQueryResult.getNonSubmittedStmts();

    writer.println("Sql file submitted. Execution ID: " + nonQueryResult.getExecutionId());
    writer.println("Submitted statements:");
    if (submittedStmts == null || submittedStmts.size() == 0) {
      writer.println("\tNone.");
    } else {
      for (String statement : submittedStmts) {
        writer.print("\t");
        writer.println(statement);
      }
    }

    if (nonsubmittedStmts != null && nonsubmittedStmts.size() != 0) {
      writer.println("Statements NOT submitted:");
      for (String statement : nonsubmittedStmts) {
        writer.print("\t");
        writer.println(statement);
      }
    }

    writer.println("Note: All query statements in a sql file are NOT submitted.");
    writer.flush();
  }

  private void commandInsertInto(CliCommand command) throws ExecutorException {
    String fullCmdStr = command.getFullCommand();
    NonQueryResult result = executor.executeNonQuery(exeContext,
        Collections.singletonList(fullCmdStr));

    writer.print("Execution submitted successfully. Id: ");
    writer.println(String.valueOf(result.getExecutionId()));
    executions.put(result.getExecutionId(), fullCmdStr);

    writer.flush();
  }

  private void commandLs(CliCommand command) {
    List<Integer> execIds = new ArrayList<>();
    String parameters = command.getParameters();
    if (CliUtil.isNullOrEmpty(parameters)) {
      execIds.addAll(executions.keySet());
    } else {
      execIds.addAll(splitExecutionIds(parameters));
    }
    if (execIds.size() == 0) {
      return;
    }

    execIds.sort(Integer::compareTo);
    final int terminalWidth = terminal.getWidth();
    final int idWidth = 3;
    final int statusWidth = 20;
    final int cmdWidth = terminalWidth - idWidth - statusWidth - 4;

    AttributedStyle oddLineStyle = AttributedStyle.DEFAULT.BOLD.foreground(AttributedStyle.BLUE);
    AttributedStyle evenLineStyle = AttributedStyle.DEFAULT.BOLD.foreground(AttributedStyle.CYAN);
    for (int i = 0; i < execIds.size(); ++i) {
      Integer id = execIds.get(i);
      String cmd = executions.get(id);
      if (cmd == null)
        continue;

      String status = "UNKNOWN";
      try {
        ExecutionStatus execStatus = executor.queryExecutionStatus(id);
        status = execStatus.name();
      } catch (ExecutorException e) {
        LOG.error("Error in commandLs: ", e);
      }

      int cmdStartIdx = 0;
      int cmdLength = cmd.length();
      StringBuilder line;
      while (cmdStartIdx < cmdLength) {
        line = new StringBuilder(terminalWidth);
        if (cmdStartIdx == 0) {
          line.append(CliConstants.SPACE);
          line.append(id);
          CliUtil.appendTo(line, 1 + idWidth + 1, CliConstants.SPACE);
          line.append(status);
        }
        CliUtil.appendTo(line, 1 + idWidth + 1 + statusWidth + 1, CliConstants.SPACE);

        int numToWrite = Math.min(cmdWidth, cmdLength - cmdStartIdx);
        if (numToWrite > 0) {
          line.append(cmd, cmdStartIdx, cmdStartIdx + numToWrite);
          cmdStartIdx += numToWrite;
        }

        if (i % 2 == 0) {
          AttributedStringBuilder attrBuilder = new AttributedStringBuilder().style(evenLineStyle);
          attrBuilder.append(line.toString());
          writer.println(attrBuilder.toAnsi());
        } else {
          AttributedStringBuilder attrBuilder = new AttributedStringBuilder().style(oddLineStyle);
          attrBuilder.append(line.toString());
          writer.println(attrBuilder.toAnsi());
        }
      }
    }
    writer.flush();
  }

  private void commandRm(CliCommand command) {
    String parameters = command.getParameters();
    List<Integer> execIds = new ArrayList<>();
    execIds.addAll(splitExecutionIds(parameters));
    for (Integer id : execIds) {
      try {
        ExecutionStatus status = executor.queryExecutionStatus(id);
        if (status == ExecutionStatus.Running) {
          writer.println(String.format("Execution %d is still running. Stop it first.", id));
          continue;
        }
        executor.removeExecution(exeContext, id);
        executions.remove(id);
      } catch (ExecutorException e) {
        writer.println("Error: " + e);
        LOG.error("Error in commandRm: ", e);
      }
    }
    writer.flush();
  }

  private void commandSelect(CliCommand command) throws ExecutorException {
    QueryResult queryResult = executor.executeQuery(exeContext, command.getFullCommand());
    CliView view = new QueryResultLogView();
    view.open(shell, queryResult);
    executor.stopExecution(exeContext, queryResult.getExecutionId());
  }

  private void commandShowTables(CliCommand command) throws ExecutorException {
    List<String> tableNames = executor.listTables(exeContext);
    for (String tableName : tableNames) {
      writer.println(tableName);
    }
    writer.flush();
  }

  private void commandShowFunctions(CliCommand command) throws ExecutorException {
    List<SqlFunction> fns = executor.listFunctions(exeContext);
    for (SqlFunction fn : fns) {
      writer.println(fn.toString());
    }
    writer.flush();
  }

  private void commandStop(CliCommand command) {
    String parameters = command.getParameters();
    if (CliUtil.isNullOrEmpty(parameters)) {
      CliUtil.printCommandUsage(command, writer);
      return;
    }

    List<Integer> execIds = new ArrayList<>();
    String[] params = parameters.split("\u0020");
    for (String param : params) {
      Integer id = null;
      try {
        id = Integer.valueOf(param);
      } catch (NumberFormatException e) {
      }
      if (id == null || !executions.containsKey(id)) {
        writer.print("Error: ");
        writer.print(param);
        writer.println(" is not a valid id.");
      } else {
        execIds.add(id);
      }
    }

    for (Integer id : execIds) {
      try {
        executor.stopExecution(exeContext, id);
        writer.println(String.format("Request to stop execution %d was sent.", id));
      } catch (ExecutorException e) {
        writer.println("Error: " + e);
        LOG.error("Error in commandStop: ", e);
      }
    }
    writer.flush();
  }

  private void commandHelp(CliCommand command) {
    String parameters = command.getParameters();
    if (CliUtil.isNullOrEmpty(parameters)) {
      printHelpMessage();
      return;
    }

    parameters = parameters.trim().toUpperCase();
    for (CliCommandType cmdType : CliCommandType.values()) {
      String cmdText = cmdType.getCommandName();
      if (cmdText.equals(parameters)) {
        writer.println(cmdType.getUsage());
        writer.flush();
        return;
      }
    }

    writer.print("Unknown command: ");
    writer.println(parameters);
    writer.flush();
  }

  private void clearScreen() {
    terminal.puts(InfoCmp.Capability.clear_screen);
  }

  /*
      Field    | Type
      -------------------------
      Field1   | Type 1
      Field2   | VARCHAR(STRING)
      Field... | VARCHAR(STRING)
      -------------------------
  */
  private List<String> formatSchema4Display(SqlSchema schema) {
    final String headerField = "Field";
    final String headerType = "Type";
    final char seperator = '|';
    final char lineSep = '-';

    int terminalWidth = terminal.getWidth();
    // Two spaces * 2 plus one SEPERATOR
    if (terminalWidth < 2 + 2 + 1 + headerField.length() + headerType.length()) {
      return Collections.singletonList("Not enough room.");
    }

    // Find the best seperator position for least rows
    int seperatorPos = headerField.length() + 2;
    int minRowNeeded = Integer.MAX_VALUE;
    int longestLineCharNum = 0;
    int rowCount = schema.getFields().size();
    for (int j = seperatorPos; j < terminalWidth - headerType.length() - 2; ++j) {
      boolean fieldWrapped = false;
      int rowNeeded = 0;
      for (int i = 0; i < rowCount; ++i) {
        SqlSchema.SqlField field = schema.getFields().get(i);
        int fieldLen = field.getFieldName().length();
        int typeLen = field.getFieldSchema().getFieldType().toString().length();
        int fieldRowNeeded = CliUtil.ceilingDiv(fieldLen, j - 2);
        int typeRowNeeded = CliUtil.ceilingDiv(typeLen, terminalWidth - 1 - j - 2);

        rowNeeded += Math.max(fieldRowNeeded, typeRowNeeded);
        fieldWrapped |= fieldRowNeeded > 1;
        if (typeRowNeeded > 1) {
          longestLineCharNum = terminalWidth;
        } else {
          longestLineCharNum = Math.max(longestLineCharNum, j + typeLen + 2 + 1);
        }
      }
      if (rowNeeded < minRowNeeded) {
        minRowNeeded = rowNeeded;
        seperatorPos = j;
      }
      if (!fieldWrapped)
        break;
    }

    List<String> lines = new ArrayList<>(minRowNeeded + 4);

    // Header
    StringBuilder line = new StringBuilder(terminalWidth);
    line.append(CliConstants.SPACE);
    line.append(headerField);
    CliUtil.appendTo(line, seperatorPos - 1, CliConstants.SPACE);
    line.append(seperator);
    line.append(CliConstants.SPACE);
    line.append(headerType);
    lines.add(line.toString());
    line = new StringBuilder(terminalWidth);
    CliUtil.appendTo(line, longestLineCharNum - 1, lineSep);
    lines.add(line.toString());

    // Body
    AttributedStyle oddLineStyle = AttributedStyle.BOLD.foreground(AttributedStyle.BLUE);
    AttributedStyle evenLineStyle = AttributedStyle.BOLD.foreground(AttributedStyle.CYAN);

    final int fieldColSize = seperatorPos - 2;
    final int typeColSize = terminalWidth - seperatorPos - 1 - 2;
    for (int i = 0; i < rowCount; ++i) {
      SqlSchema.SqlField sqlField = schema.getFields().get(i);
      String field = sqlField.getFieldName();
      String type = getFieldDisplayValue(sqlField.getFieldSchema());
      int fieldLen = field.length();
      int typeLen = type.length();
      int fieldStartIdx = 0, typeStartIdx = 0;
      while (fieldStartIdx < fieldLen || typeStartIdx < typeLen) {
        line = new StringBuilder(terminalWidth);
        line.append(CliConstants.SPACE);
        int numToWrite = Math.min(fieldColSize, fieldLen - fieldStartIdx);
        if (numToWrite > 0) {
          line.append(field, fieldStartIdx, fieldStartIdx + numToWrite);
          fieldStartIdx += numToWrite;
        }
        CliUtil.appendTo(line, seperatorPos - 1, CliConstants.SPACE);
        line.append(seperator);
        line.append(CliConstants.SPACE);

        numToWrite = Math.min(typeColSize, typeLen - typeStartIdx);
        if (numToWrite > 0) {
          line.append(type, typeStartIdx, typeStartIdx + numToWrite);
          typeStartIdx += numToWrite;
        }

        if (i % 2 == 0) {
          AttributedStringBuilder attrBuilder = new AttributedStringBuilder().style(evenLineStyle);
          attrBuilder.append(line.toString());
          lines.add(attrBuilder.toAnsi());
        } else {
          AttributedStringBuilder attrBuilder = new AttributedStringBuilder().style(oddLineStyle);
          attrBuilder.append(line.toString());
          lines.add(attrBuilder.toAnsi());
        }
      }
    }

    // Footer
    line = new StringBuilder(terminalWidth);
    CliUtil.appendTo(line, longestLineCharNum - 1, lineSep);
    lines.add(line.toString());
    return lines;
  }

  private String getFieldDisplayValue(SqlFieldSchema fieldSchema) {
    if (!isComplexField(fieldSchema.getFieldType())) {
      return fieldSchema.getFieldType().toString();
    }
    SamzaSqlFieldType fieldType = fieldSchema.getFieldType();
    switch (fieldType) {
      case ARRAY:
        return String.format("ARRAY(%s)", getFieldDisplayValue(fieldSchema.getElementSchema()));
      case MAP:
        return String.format("MAP(%s, %s)", SamzaSqlFieldType.STRING.toString(),
            getFieldDisplayValue(fieldSchema.getValueSchema()));
      case ROW:
        String rowDisplayValue = fieldSchema.getRowSchema()
            .getFields()
            .stream()
            .map(f -> getFieldDisplayValue(f.getFieldSchema()))
            .collect(Collectors.joining(","));
        return String.format("ROW(%s)", rowDisplayValue);
      default:
        throw new UnsupportedOperationException("Unknown field type " + fieldType);
    }
  }

  private boolean isComplexField(SamzaSqlFieldType fieldtype) {
    return fieldtype == SamzaSqlFieldType.ARRAY || fieldtype == SamzaSqlFieldType.MAP
        || fieldtype == SamzaSqlFieldType.ROW;
  }

  private List<Integer> splitExecutionIds(String parameters) {
    List<Integer> execIds = new ArrayList<>();
    String[] params = parameters.split("\u0020");
    for (String param : params) {
      Integer id = null;
      try {
        id = Integer.valueOf(param);
      } catch (NumberFormatException e) {
      }
      if (id == null || !executions.containsKey(id)) {
        writer.print("Error: ");
        writer.print(param);
        writer.println(" is not a valid id.");
      } else {
        execIds.add(id);
      }
    }
    return execIds;
  }
}