/*
 * 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.ambari.server.upgrade;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.audit.AuditLoggerModule;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.ControllerModule;
import org.apache.ambari.server.ldap.LdapModule;
import org.apache.ambari.server.orm.DBAccessor;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.utils.EventBusSynchronizer;
import org.apache.ambari.server.utils.VersionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.support.JdbcUtils;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.inject.Guice;
import com.google.inject.Inject;
import com.google.inject.Injector;
import com.google.inject.multibindings.Multibinder;
import com.google.inject.persist.PersistService;

public class SchemaUpgradeHelper {
  private static final Logger LOG = LoggerFactory.getLogger
    (SchemaUpgradeHelper.class);

  private Set<UpgradeCatalog> allUpgradeCatalogs;
  private PersistService persistService;
  private DBAccessor dbAccessor;
  private Configuration configuration;
  private static final String[] rcaTableNames = {"workflow", "job", "task", "taskAttempt", "hdfsEvent", "mapreduceEvent", "clusterEvent"};
  static final Gson gson = new GsonBuilder().create();

  @Inject
  public SchemaUpgradeHelper(Set<UpgradeCatalog> allUpgradeCatalogs,
                             PersistService persistService,
                             DBAccessor dbAccessor,
                             Configuration configuration) {
    this.allUpgradeCatalogs = allUpgradeCatalogs;
    this.persistService = persistService;
    this.dbAccessor = dbAccessor;
    this.configuration = configuration;
  }

  public void startPersistenceService() {
    persistService.start();
  }

  public void stopPersistenceService() {
    persistService.stop();
  }

  public Set<UpgradeCatalog> getAllUpgradeCatalogs() {
    return allUpgradeCatalogs;
  }

  public String readSourceVersion() {

    Statement statement = null;
    ResultSet rs = null;
    try {
      statement = dbAccessor.getConnection().createStatement();
      if (statement != null) {
        rs = statement.executeQuery("SELECT " + dbAccessor.quoteObjectName("metainfo_value") +
          " from metainfo WHERE " + dbAccessor.quoteObjectName("metainfo_key") + "='version'");
        if (rs != null && rs.next()) {
          return rs.getString(1);
        }
      }
    } catch (SQLException e) {
      throw new RuntimeException("Unable to read database version", e);

    } finally {
      if (rs != null) {
        try {
          rs.close();
        } catch (SQLException e) {
          throw new RuntimeException("Cannot close result set");
        }
      }
      if (statement != null) {
        try {
          statement.close();
        } catch (SQLException e) {
          throw new RuntimeException("Cannot close statement");
        }
      }
    }
    //not found, assume oldest version
    //doesn't matter as there single upgrade catalog for 1.2.0 - 1.5.0 and 1.4.4 - 1.5.0 upgrades
    return "1.2.0";
  }

  /**
   * Read server version file
   * @return
   */
  protected String getAmbariServerVersion() {
    return configuration.getServerVersion();
  }

  /**
   * Return a set Upgrade catalogs to be applied to upgrade from
   * @sourceVersion to @targetVersion
   *
   * @param sourceVersion
   * @param targetVersion
   * @return
   * @throws org.apache.ambari.server.AmbariException
   */
  protected List<UpgradeCatalog> getUpgradePath(String sourceVersion,
                                                       String targetVersion) throws AmbariException {
    List<UpgradeCatalog> upgradeCatalogs = new ArrayList<>();
    List<UpgradeCatalog> candidateCatalogs = new ArrayList<>(allUpgradeCatalogs);

    Collections.sort(candidateCatalogs, new AbstractUpgradeCatalog.VersionComparator());

    for (UpgradeCatalog upgradeCatalog : candidateCatalogs) {
      if (sourceVersion == null || VersionUtils.compareVersions(sourceVersion,
        upgradeCatalog.getTargetVersion(), 4) < 0) {
        // catalog version is newer than source
        if (VersionUtils.compareVersions(upgradeCatalog.getTargetVersion(),
          targetVersion, 4) <= 0) {
          // catalog version is older or equal to target
          upgradeCatalogs.add(upgradeCatalog);
        }
      }
    }

    LOG.info("Upgrade path: " + upgradeCatalogs);

    return upgradeCatalogs;
  }

  /**
   * Extension of main controller module
   */
  public static class UpgradeHelperModule extends ControllerModule {

    public UpgradeHelperModule() throws Exception {
    }

    public UpgradeHelperModule(Properties properties) throws Exception {
      super(properties);
    }

    @Override
    protected void configure() {
      super.configure();
      // Add binding to each newly created catalog
      Multibinder<UpgradeCatalog> catalogBinder = Multibinder.newSetBinder(binder(), UpgradeCatalog.class);
      catalogBinder.addBinding().to(UpgradeCatalog251.class);
      catalogBinder.addBinding().to(UpgradeCatalog252.class);
      catalogBinder.addBinding().to(UpgradeCatalog260.class);
      catalogBinder.addBinding().to(UpgradeCatalog261.class);
      catalogBinder.addBinding().to(UpgradeCatalog262.class);
      catalogBinder.addBinding().to(UpgradeCatalog270.class);
      catalogBinder.addBinding().to(UpgradeCatalog271.class);
      catalogBinder.addBinding().to(UpgradeCatalog272.class);
      catalogBinder.addBinding().to(UpgradeCatalog274.class);
      catalogBinder.addBinding().to(UpgradeCatalog275.class);
      catalogBinder.addBinding().to(UpgradeCatalog276.class);
      catalogBinder.addBinding().to(UpgradeCatalog280.class);
      catalogBinder.addBinding().to(UpdateAlertScriptPaths.class);
      catalogBinder.addBinding().to(FinalUpgradeCatalog.class);

      EventBusSynchronizer.synchronizeAmbariEventPublisher(binder());
    }
  }

  public void executeUpgrade(List<UpgradeCatalog> upgradeCatalogs) throws AmbariException {
    LOG.info("Executing DDL upgrade...");

    if (upgradeCatalogs != null && !upgradeCatalogs.isEmpty()) {
      for (UpgradeCatalog upgradeCatalog : upgradeCatalogs) {
        try {
          upgradeCatalog.upgradeSchema();
        } catch (Exception e) {
          LOG.error("Upgrade failed. ", e);
          throw new AmbariException(e.getMessage(), e);
        }
      }
    }
  }

  public void executePreDMLUpdates(List<UpgradeCatalog> upgradeCatalogs) throws AmbariException {
    LOG.info("Executing Pre-DML changes.");

    if (upgradeCatalogs != null && !upgradeCatalogs.isEmpty()) {
      for (UpgradeCatalog upgradeCatalog : upgradeCatalogs) {
        try {
          upgradeCatalog.preUpgradeData();
        } catch (Exception e) {
          LOG.error("Upgrade failed. ", e);
          throw new AmbariException(e.getMessage(), e);
        }
      }
    }
  }

  public void executeDMLUpdates(List<UpgradeCatalog> upgradeCatalogs, String ambariUpgradeConfigUpdatesFileName) throws AmbariException {
    LOG.info("Executing DML changes.");

    if (upgradeCatalogs != null && !upgradeCatalogs.isEmpty()) {
      for (UpgradeCatalog upgradeCatalog : upgradeCatalogs) {
        try {
          upgradeCatalog.setConfigUpdatesFileName(ambariUpgradeConfigUpdatesFileName);
          upgradeCatalog.upgradeData();
        } catch (Exception e) {
          LOG.error("Upgrade failed. ", e);
          throw new AmbariException(e.getMessage(), e);
        }
      }
    }
  }

  public void executeOnPostUpgrade(List<UpgradeCatalog> upgradeCatalogs)
      throws AmbariException {
    LOG.info("Finalizing catalog upgrade.");

    if (upgradeCatalogs != null && !upgradeCatalogs.isEmpty()) {
      for (UpgradeCatalog upgradeCatalog : upgradeCatalogs) {
        try {
          upgradeCatalog.onPostUpgrade();
          upgradeCatalog.updateDatabaseSchemaVersion();
        } catch (Exception e) {
          LOG.error("Upgrade failed. ", e);
          throw new AmbariException(e.getMessage(), e);
        }
      }
    }
  }

  public void outputUpgradeJsonOutput(List<UpgradeCatalog> upgradeCatalogs)
      throws AmbariException {
    LOG.info("Combining upgrade json output.");
    Map<String,String> combinedUpgradeJsonOutput = new HashMap<>();

    if (upgradeCatalogs != null && !upgradeCatalogs.isEmpty()) {
      for (UpgradeCatalog upgradeCatalog : upgradeCatalogs) {
        try {
          combinedUpgradeJsonOutput.putAll(upgradeCatalog.getUpgradeJsonOutput());

        } catch (Exception e) {
          LOG.error("Upgrade failed. ", e);
          throw new AmbariException(e.getMessage(), e);
        }
      }
    }
    String content = gson.toJson(combinedUpgradeJsonOutput);
    System.out.println(content);
  }

  public void resetUIState() throws AmbariException {
    LOG.info("Resetting UI state.");
    try {
      dbAccessor.updateTable("key_value_store", dbAccessor.quoteObjectName("value"),
          "{\"clusterState\":\"CLUSTER_STARTED_5\"}",
          "where " + dbAccessor.quoteObjectName("key") + "='CLUSTER_CURRENT_STATUS'");
    } catch (SQLException e) {
      throw new AmbariException("Unable to reset UI state", e);
    }
  }

  public void cleanUpRCATables() {
    LOG.info("Cleaning up RCA tables.");
    for (String tableName : rcaTableNames) {
      try {
        if (dbAccessor.tableExists(tableName)) {
          dbAccessor.truncateTable(tableName);
        }
      } catch (Exception e) {
        LOG.warn("Error cleaning rca table " + tableName, e);
      }
    }
    try {
      cleanUpTablesFromRCADatabase();
    } catch (Exception e) {
      LOG.warn("Error cleaning rca tables from ambarirca db", e);
    }
  }

  private void cleanUpTablesFromRCADatabase() throws ClassNotFoundException, SQLException {
    String driverName = configuration.getRcaDatabaseDriver();
    String connectionURL = configuration.getRcaDatabaseUrl();
    if (connectionURL.contains(Configuration.HOSTNAME_MACRO)) {
      connectionURL = connectionURL.replace(Configuration.HOSTNAME_MACRO, "localhost");
    }
    String username = configuration.getRcaDatabaseUser();
    String password = configuration.getRcaDatabasePassword();

    Class.forName(driverName);
    try (Connection connection = DriverManager.getConnection(connectionURL, username, password)) {
      connection.setAutoCommit(true);
      for (String tableName : rcaTableNames) {
        String query = "DELETE FROM " + tableName;
        try (Statement statement = connection.createStatement()) {
          statement.execute(query);
        } catch (Exception e) {
          LOG.warn("Error while executing query: " + query, e);
        }
      }
    }
  }

  /**
   * Returns minimal version of available {@link UpgradeCatalog}
   *
   * @return string representation of minimal version of {@link UpgradeCatalog}
   */
  private String getMinimalUpgradeCatalogVersion(){
    List<UpgradeCatalog> candidateCatalogs = new ArrayList<>(allUpgradeCatalogs);
    Collections.sort(candidateCatalogs, new AbstractUpgradeCatalog.VersionComparator());

    if (candidateCatalogs.isEmpty()) {
      return null;
    }

    return candidateCatalogs.iterator().next().getTargetVersion();
  }

  /**
   * Checks if source version meets minimal requirements for upgrade
   *
   * @param minUpgradeVersion min allowed version for the upgrade, could be obtained via {@link #getMinimalUpgradeCatalogVersion()}
   * @param sourceVersion current version of the Database, which need to be upgraded
   *
   * @return  true if upgrade is allowed or false if not
   */
  private boolean verifyUpgradePath(String minUpgradeVersion, String sourceVersion){
    if (null == minUpgradeVersion){
      return false;
    }

    return VersionUtils.compareVersions(sourceVersion, minUpgradeVersion) >= 0;
  }

  private List<String> getMyISAMTables() throws SQLException {
    if (!configuration.getDatabaseType().equals(Configuration.DatabaseType.MYSQL)) {
      return Collections.emptyList();
    }
    List<String> myISAMTables = new ArrayList<>();
    String query = String.format("SELECT table_name FROM information_schema.tables WHERE table_schema = '%s' " +
      "AND engine = 'MyISAM' AND table_type = 'BASE TABLE'", configuration.getServerDBName());
    Statement statement = null;
    ResultSet rs = null;
    try {
      statement = dbAccessor.getConnection().createStatement();
      rs = statement.executeQuery(query);
      if (rs != null) {
        while (rs.next()) {
          myISAMTables.add(rs.getString("table_name"));
        }
      }
    } finally {
      JdbcUtils.closeResultSet(rs);
      JdbcUtils.closeStatement(statement);
    }
    return myISAMTables;
  }

  /**
   * Upgrade Ambari DB schema to the target version passed in as the only
   * argument.
   * @param args args[0] = target version to upgrade to.
   */
  public static void main(String[] args) throws Exception {
    try {
      // check java version to be higher then 1.6
      String[] splittedJavaVersion = System.getProperty("java.version").split("\\.");
      float javaVersion = Float.parseFloat(splittedJavaVersion[0] + "." + splittedJavaVersion[1]);
      if (javaVersion < Configuration.JDK_MIN_VERSION) {
        LOG.error(String.format("Oracle JDK version is lower than %.1f It can cause problems during upgrade process. Please," +
                " use 'ambari-server setup' command to upgrade JDK!", Configuration.JDK_MIN_VERSION));
        System.exit(1);
      }

      Injector injector = Guice.createInjector(new UpgradeHelperModule(), new AuditLoggerModule(), new LdapModule());

      // Startup the JPA infrastructure, but do not indicate it is initialized since the underlying
      // database schema may not be updated to meet the expectations of the Entity instances.
      GuiceJpaInitializer jpaInitializer = injector.getInstance(GuiceJpaInitializer.class);

      SchemaUpgradeHelper schemaUpgradeHelper = injector.getInstance(SchemaUpgradeHelper.class);

      //Fail if MySQL database has tables with MyISAM engine
      List<String> myISAMTables = schemaUpgradeHelper.getMyISAMTables();
      if (!myISAMTables.isEmpty()) {
        String errorMessage = String.format("Unsupported MyISAM table %s detected. " +
            "For correct upgrade database should be migrated to InnoDB engine.", myISAMTables.get(0));
        LOG.error(errorMessage);
        throw new AmbariException(errorMessage);
      }

      String targetVersion = schemaUpgradeHelper.getAmbariServerVersion();
      LOG.info("Upgrading schema to target version = " + targetVersion);

      UpgradeCatalog targetUpgradeCatalog = AbstractUpgradeCatalog
        .getUpgradeCatalog(targetVersion);

      LOG.debug("Target upgrade catalog. {}", targetUpgradeCatalog);

      // Read source version from DB
      String sourceVersion = schemaUpgradeHelper.readSourceVersion();
      LOG.info("Upgrading schema from source version = " + sourceVersion);

      String minimalRequiredUpgradeVersion = schemaUpgradeHelper.getMinimalUpgradeCatalogVersion();

      if (!schemaUpgradeHelper.verifyUpgradePath(minimalRequiredUpgradeVersion, sourceVersion)){
        throw new AmbariException(String.format("Database version does not meet minimal upgrade requirements. Expected version should be not less than %s, current version is %s",
          minimalRequiredUpgradeVersion, sourceVersion));
      }

      List<UpgradeCatalog> upgradeCatalogs =
        schemaUpgradeHelper.getUpgradePath(sourceVersion, targetVersion);

      String date = new SimpleDateFormat("MM-dd-yyyy_HH:mm:ss").format(new Date());
      String ambariUpgradeConfigUpdatesFileName = "ambari_upgrade_config_changes_" + date + ".json";

      schemaUpgradeHelper.executeUpgrade(upgradeCatalogs);

      // The DDL is expected to be updated, now send the JPA initialized event so Entity
      // implementations can be created.
      jpaInitializer.setInitialized();

      schemaUpgradeHelper.executePreDMLUpdates(upgradeCatalogs);

      schemaUpgradeHelper.executeDMLUpdates(upgradeCatalogs, ambariUpgradeConfigUpdatesFileName);

      schemaUpgradeHelper.executeOnPostUpgrade(upgradeCatalogs);
      schemaUpgradeHelper.outputUpgradeJsonOutput(upgradeCatalogs);

      schemaUpgradeHelper.resetUIState();

      LOG.info("Upgrade successful.");

      schemaUpgradeHelper.cleanUpRCATables();

      schemaUpgradeHelper.stopPersistenceService();

      // Signal all threads that we are ready to exit...
      System.exit(0);
    } catch (Throwable e) {
      if (e instanceof AmbariException) {
        LOG.error("Exception occurred during upgrade, failed", e);
        throw (AmbariException)e;
      }else{
        LOG.error("Unexpected error, upgrade failed", e);
        throw new Exception("Unexpected error, upgrade failed", e);
      }
    }
  }
}
