/*
 * Copyright 2012 LinkedIn Corp.
 *
 * 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 azkaban.trigger;

import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.commons.dbutils.DbUtils;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.ResultSetHandler;
import org.apache.log4j.Logger;

import org.joda.time.DateTime;

import azkaban.database.AbstractJdbcLoader;
import azkaban.utils.GZIPUtils;
import azkaban.utils.JSONUtils;
import azkaban.utils.Props;

public class JdbcTriggerLoader extends AbstractJdbcLoader implements
    TriggerLoader {
  private static Logger logger = Logger.getLogger(JdbcTriggerLoader.class);

  private EncodingType defaultEncodingType = EncodingType.GZIP;

  private static final String triggerTblName = "triggers";

  private static final String GET_UPDATED_TRIGGERS =
      "SELECT trigger_id, trigger_source, modify_time, enc_type, data FROM "
          + triggerTblName + " WHERE modify_time>=?";

  private static String GET_ALL_TRIGGERS =
      "SELECT trigger_id, trigger_source, modify_time, enc_type, data FROM "
          + triggerTblName;

  private static String GET_TRIGGER =
      "SELECT trigger_id, trigger_source, modify_time, enc_type, data FROM "
          + triggerTblName + " WHERE trigger_id=?";

  private static String ADD_TRIGGER = "INSERT INTO " + triggerTblName
      + " ( modify_time) values (?)";

  private static String REMOVE_TRIGGER = "DELETE FROM " + triggerTblName
      + " WHERE trigger_id=?";

  private static String UPDATE_TRIGGER =
      "UPDATE "
          + triggerTblName
          + " SET trigger_source=?, modify_time=?, enc_type=?, data=? WHERE trigger_id=?";

  public EncodingType getDefaultEncodingType() {
    return defaultEncodingType;
  }

  public void setDefaultEncodingType(EncodingType defaultEncodingType) {
    this.defaultEncodingType = defaultEncodingType;
  }

  public JdbcTriggerLoader(Props props) {
    super(props);
  }

  @Override
  public List<Trigger> getUpdatedTriggers(long lastUpdateTime)
      throws TriggerLoaderException {
    logger.info("Loading triggers changed since "
        + new DateTime(lastUpdateTime).toString());
    Connection connection = getConnection();

    QueryRunner runner = new QueryRunner();
    ResultSetHandler<List<Trigger>> handler = new TriggerResultHandler();

    List<Trigger> triggers;

    try {
      triggers =
          runner.query(connection, GET_UPDATED_TRIGGERS, handler,
              lastUpdateTime);
    } catch (SQLException e) {
      logger.error(GET_ALL_TRIGGERS + " failed.");

      throw new TriggerLoaderException("Loading triggers from db failed. ", e);
    } finally {
      DbUtils.closeQuietly(connection);
    }

    logger.info("Loaded " + triggers.size() + " triggers.");

    return triggers;
  }

  @Override
  public List<Trigger> loadTriggers() throws TriggerLoaderException {
    logger.info("Loading all triggers from db.");
    Connection connection = getConnection();

    QueryRunner runner = new QueryRunner();
    ResultSetHandler<List<Trigger>> handler = new TriggerResultHandler();

    List<Trigger> triggers;

    try {
      triggers = runner.query(connection, GET_ALL_TRIGGERS, handler);
    } catch (SQLException e) {
      logger.error(GET_ALL_TRIGGERS + " failed.");

      throw new TriggerLoaderException("Loading triggers from db failed. ", e);
    } finally {
      DbUtils.closeQuietly(connection);
    }

    logger.info("Loaded " + triggers.size() + " triggers.");

    return triggers;
  }

  @Override
  public void removeTrigger(Trigger t) throws TriggerLoaderException {
    logger.info("Removing trigger " + t.toString() + " from db.");

    QueryRunner runner = createQueryRunner();
    try {
      int removes = runner.update(REMOVE_TRIGGER, t.getTriggerId());
      if (removes == 0) {
        throw new TriggerLoaderException("No trigger has been removed.");
      }
    } catch (SQLException e) {
      logger.error(REMOVE_TRIGGER + " failed.");
      throw new TriggerLoaderException("Remove trigger " + t.toString()
          + " from db failed. ", e);
    }
  }

  @Override
  public void addTrigger(Trigger t) throws TriggerLoaderException {
    logger.info("Inserting trigger " + t.toString() + " into db.");
    t.setLastModifyTime(System.currentTimeMillis());
    Connection connection = getConnection();
    try {
      addTrigger(connection, t, defaultEncodingType);
    } catch (Exception e) {
      throw new TriggerLoaderException("Error uploading trigger", e);
    } finally {
      DbUtils.closeQuietly(connection);
    }
  }

  private synchronized void addTrigger(Connection connection, Trigger t,
      EncodingType encType) throws TriggerLoaderException {

    QueryRunner runner = new QueryRunner();

    long id;

    try {
      runner.update(connection, ADD_TRIGGER, DateTime.now().getMillis());
      connection.commit();
      id =
          runner.query(connection, LastInsertID.LAST_INSERT_ID,
              new LastInsertID());

      if (id == -1l) {
        logger.error("trigger id is not properly created.");
        throw new TriggerLoaderException("trigger id is not properly created.");
      }

      t.setTriggerId((int) id);
      updateTrigger(t);
      logger.info("uploaded trigger " + t.getDescription());
    } catch (SQLException e) {
      throw new TriggerLoaderException("Error creating trigger.", e);
    }

  }

  @Override
  public void updateTrigger(Trigger t) throws TriggerLoaderException {
    logger.info("Updating trigger " + t.getTriggerId() + " into db.");
    t.setLastModifyTime(System.currentTimeMillis());
    Connection connection = getConnection();
    try {
      updateTrigger(connection, t, defaultEncodingType);
    } catch (Exception e) {
      e.printStackTrace();
      throw new TriggerLoaderException("Failed to update trigger "
          + t.toString() + " into db!");
    } finally {
      DbUtils.closeQuietly(connection);
    }
  }

  private void updateTrigger(Connection connection, Trigger t,
      EncodingType encType) throws TriggerLoaderException {

    String json = JSONUtils.toJSON(t.toJson());
    byte[] data = null;
    try {
      byte[] stringData = json.getBytes("UTF-8");
      data = stringData;

      if (encType == EncodingType.GZIP) {
        data = GZIPUtils.gzipBytes(stringData);
      }
      logger.debug("NumChars: " + json.length() + " UTF-8:" + stringData.length
          + " Gzip:" + data.length);
    } catch (IOException e) {
      throw new TriggerLoaderException("Error encoding the trigger "
          + t.toString());
    }

    QueryRunner runner = new QueryRunner();

    try {
      int updates =
          runner.update(connection, UPDATE_TRIGGER, t.getSource(),
              t.getLastModifyTime(), encType.getNumVal(), data,
              t.getTriggerId());
      connection.commit();
      if (updates == 0) {
        throw new TriggerLoaderException("No trigger has been updated.");
        // logger.error("No trigger is updated!");
      } else {
        logger.info("Updated " + updates + " records.");
      }
    } catch (SQLException e) {
      logger.error(UPDATE_TRIGGER + " failed.");
      throw new TriggerLoaderException("Update trigger " + t.toString()
          + " into db failed. ", e);
    }
  }

  private static class LastInsertID implements ResultSetHandler<Long> {
    private static String LAST_INSERT_ID = "SELECT LAST_INSERT_ID()";

    @Override
    public Long handle(ResultSet rs) throws SQLException {
      if (!rs.next()) {
        return -1l;
      }

      long id = rs.getLong(1);
      return id;
    }

  }

  public class TriggerResultHandler implements ResultSetHandler<List<Trigger>> {

    @Override
    public List<Trigger> handle(ResultSet rs) throws SQLException {
      if (!rs.next()) {
        return Collections.<Trigger> emptyList();
      }

      ArrayList<Trigger> triggers = new ArrayList<Trigger>();
      do {
        int triggerId = rs.getInt(1);
        // String triggerSource = rs.getString(2);
        // long modifyTime = rs.getLong(3);
        int encodingType = rs.getInt(4);
        byte[] data = rs.getBytes(5);

        Object jsonObj = null;
        if (data != null) {
          EncodingType encType = EncodingType.fromInteger(encodingType);

          try {
            // Convoluted way to inflate strings. Should find common package or
            // helper function.
            if (encType == EncodingType.GZIP) {
              // Decompress the sucker.
              String jsonString = GZIPUtils.unGzipString(data, "UTF-8");
              jsonObj = JSONUtils.parseJSONFromString(jsonString);
            } else {
              String jsonString = new String(data, "UTF-8");
              jsonObj = JSONUtils.parseJSONFromString(jsonString);
            }
          } catch (IOException e) {
            throw new SQLException("Error reconstructing trigger data ");
          }
        }

        Trigger t = null;
        try {
          t = Trigger.fromJson(jsonObj);
          triggers.add(t);
        } catch (Exception e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
          logger.error("Failed to load trigger " + triggerId);
        }
      } while (rs.next());

      return triggers;
    }

  }

  private Connection getConnection() throws TriggerLoaderException {
    Connection connection = null;
    try {
      connection = super.getDBConnection(false);
    } catch (Exception e) {
      DbUtils.closeQuietly(connection);
      throw new TriggerLoaderException("Error getting DB connection.", e);
    }

    return connection;
  }

  @Override
  public Trigger loadTrigger(int triggerId) throws TriggerLoaderException {
    logger.info("Loading trigger " + triggerId + " from db.");
    Connection connection = getConnection();

    QueryRunner runner = new QueryRunner();
    ResultSetHandler<List<Trigger>> handler = new TriggerResultHandler();

    List<Trigger> triggers;

    try {
      triggers = runner.query(connection, GET_TRIGGER, handler, triggerId);
    } catch (SQLException e) {
      logger.error(GET_TRIGGER + " failed.");
      throw new TriggerLoaderException("Loading trigger from db failed. ", e);
    } finally {
      DbUtils.closeQuietly(connection);
    }

    if (triggers.size() == 0) {
      logger.error("Loaded 0 triggers. Failed to load trigger " + triggerId);
      throw new TriggerLoaderException(
          "Loaded 0 triggers. Failed to load trigger " + triggerId);
    }

    return triggers.get(0);
  }

}
