// Copyright 2008 Google Inc.
//
// 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 com.google.gwtorm.jdbc;

import com.google.common.base.Preconditions;
import com.google.gwtorm.schema.ColumnModel;
import com.google.gwtorm.schema.RelationModel;
import com.google.gwtorm.schema.SchemaModel;
import com.google.gwtorm.schema.SequenceModel;
import com.google.gwtorm.schema.sql.SqlDialect;
import com.google.gwtorm.server.AbstractSchema;
import com.google.gwtorm.server.OrmConcurrencyException;
import com.google.gwtorm.server.OrmDuplicateKeyException;
import com.google.gwtorm.server.OrmException;
import com.google.gwtorm.server.Schema;
import com.google.gwtorm.server.StatementExecutor;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.HashSet;
import java.util.Set;

/** Internal base class for implementations of {@link Schema}. */
public abstract class JdbcSchema extends AbstractSchema {
  private final Database<?> dbDef;
  private Connection conn;
  private OrmException transactionException;

  protected JdbcSchema(final Database<?> d) throws OrmException {
    dbDef = d;
    conn = dbDef.newConnection();
  }

  public final Connection getConnection() {
    return conn;
  }

  public final SqlDialect getDialect() {
    return dbDef.getDialect();
  }

  @Override
  public void commit() throws OrmException {
    try {
      if (isInTransaction()) {
        if (transactionException != null) {
          OrmException e = transactionException;
          transactionException = null;
          if (e instanceof OrmConcurrencyException) {
            throw new OrmConcurrencyException(e.getMessage(), e);
          } else if (e instanceof OrmDuplicateKeyException) {
            throw new OrmDuplicateKeyException(e.getMessage(), e);
          } else {
            throw new OrmException(e.getMessage(), e);
          }
        }
        conn.commit();
      }
    } catch (SQLException err) {
      throw new OrmException("Cannot commit transaction", err);
    } finally {
      try {
        conn.setAutoCommit(true);
      } catch (SQLException err) {
        throw new OrmException("Cannot set auto commit mode", err);
      }
    }
  }

  @Override
  public void rollback() throws OrmException {
    try {
      if (!conn.getAutoCommit()) {
        transactionException = null;
        conn.rollback();
      }
    } catch (SQLException err) {
      throw new OrmException("Cannot rollback transaction", err);
    } finally {
      try {
        conn.setAutoCommit(true);
      } catch (SQLException err) {
        throw new OrmException("Cannot set auto commit mode", err);
      }
    }
  }

  @Override
  public void updateSchema(final StatementExecutor e) throws OrmException {
    try {
      createSequences(e);
      createRelations(e);

      for (final RelationModel rel : dbDef.getSchemaModel().getRelations()) {
        addColumns(e, rel);
      }
    } catch (SQLException err) {
      throw new OrmException("Cannot update schema", err);
    }
  }

  private void createSequences(final StatementExecutor e) throws OrmException,
      SQLException {
    final SqlDialect dialect = dbDef.getDialect();
    final SchemaModel model = dbDef.getSchemaModel();

    Set<String> have = dialect.listSequences(getConnection());
    for (final SequenceModel s : model.getSequences()) {
      if (!have.contains(s.getSequenceName().toLowerCase())) {
        e.execute(s.getCreateSequenceSql(dialect));
      }
    }
  }

  private void createRelations(final StatementExecutor e) throws SQLException,
      OrmException {
    final SqlDialect dialect = dbDef.getDialect();
    final SchemaModel model = dbDef.getSchemaModel();
    Set<String> have = dialect.listTables(getConnection());
    for (final RelationModel r : model.getRelations()) {
      if (!have.contains(r.getRelationName().toLowerCase())) {
        e.execute(r.getCreateTableSql(dialect));
      }
    }
  }

  private void addColumns(final StatementExecutor e, final RelationModel rel)
      throws SQLException, OrmException {
    final SqlDialect dialect = dbDef.getDialect();
    Set<String> have = dialect.listColumns( //
        getConnection(), rel.getRelationName().toLowerCase());
    for (final ColumnModel c : rel.getColumns()) {
      if (!have.contains(c.getColumnName().toLowerCase())) {
        dialect.addColumn(e, rel.getRelationName(), c);
      }
    }
  }

  public void renameTable(final StatementExecutor e, String from, String to)
      throws OrmException {
    Preconditions.checkNotNull(e);
    Preconditions.checkNotNull(from);
    Preconditions.checkNotNull(to);
    getDialect().renameTable(e, from, to);
  }

  public void renameField(final StatementExecutor e, String table, String from,
      String to) throws OrmException {
    final RelationModel rel = findRelationModel(table);
    if (rel == null) {
      throw new OrmException("Relation " + table + " not defined");
    }
    final ColumnModel col = rel.getField(to);
    if (col == null) {
      throw new OrmException("Relation " + table + " does not have " + to);
    }
    getDialect().renameColumn(e, table, from, col);
  }

  public void renameColumn(final StatementExecutor e, String table, String from,
      String to) throws OrmException {
    final RelationModel rel = findRelationModel(table);
    if (rel == null) {
      throw new OrmException("Relation " + table + " not defined");
    }
    final ColumnModel col = rel.getColumn(to);
    if (col == null) {
      throw new OrmException("Relation " + table + " does not have " + to);
    }
    getDialect().renameColumn(e, table, from, col);
  }

  private RelationModel findRelationModel(String table) {
    for (final RelationModel rel : dbDef.getSchemaModel().getRelations()) {
      if (table.equalsIgnoreCase(rel.getRelationName())) {
        return rel;
      }
    }
    return null;
  }

  @Override
  public void pruneSchema(final StatementExecutor e) throws OrmException {
    try {
      pruneSequences(e);
      pruneRelations(e);

      for (final RelationModel rel : dbDef.getSchemaModel().getRelations()) {
        pruneColumns(e, rel);
      }
    } catch (SQLException err) {
      throw new OrmException("Schema prune failure", err);
    }
  }

  private void pruneSequences(final StatementExecutor e) throws SQLException,
      OrmException {
    final SqlDialect dialect = dbDef.getDialect();
    final SchemaModel model = dbDef.getSchemaModel();
    HashSet<String> want = new HashSet<>();
    for (final SequenceModel s : model.getSequences()) {
      want.add(s.getSequenceName().toLowerCase());
    }
    for (final String sequence : dialect.listSequences(getConnection())) {
      if (!want.contains(sequence)) {
        e.execute(dialect.getDropSequenceSql(sequence));
      }
    }
  }

  private void pruneRelations(final StatementExecutor e) throws SQLException,
      OrmException {
    final SqlDialect dialect = dbDef.getDialect();
    final SchemaModel model = dbDef.getSchemaModel();
    HashSet<String> want = new HashSet<>();
    for (final RelationModel r : model.getRelations()) {
      want.add(r.getRelationName().toLowerCase());
    }
    for (final String table : dialect.listTables(getConnection())) {
      if (!want.contains(table)) {
        e.execute("DROP TABLE " + table);
      }
    }
  }

  private void pruneColumns(final StatementExecutor e, final RelationModel rel)
      throws SQLException, OrmException {
    final SqlDialect dialect = dbDef.getDialect();
    HashSet<String> want = new HashSet<>();
    for (final ColumnModel c : rel.getColumns()) {
      want.add(c.getColumnName().toLowerCase());
    }
    for (String column : dialect.listColumns( //
        getConnection(), rel.getRelationName().toLowerCase())) {
      if (!want.contains(column)) {
        dialect.dropColumn(e, rel.getRelationName(), column);
      }
    }
  }

  @Override
  protected long nextLong(final String poolName) throws OrmException {
    return getDialect().nextLong(getConnection(), poolName);
  }

  @Override
  public void close() {
    transactionException = null;
    if (conn != null) {
      try {
        conn.close();
      } catch (SQLException err) {
        // TODO Handle an exception while closing a connection
      }
      conn = null;
    }
  }

  boolean isInTransaction() throws SQLException {
    return !conn.getAutoCommit();
  }

  void setTransactionException(OrmException ex) {
    // commit() needs a single cause, so just take the first.
    if (transactionException == null) {
      transactionException = Preconditions.checkNotNull(ex);
    }
  }
}