/*
    * Copyright 2006 Outsource Cafe, 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 org.javagen.agile.db;
  
  import java.sql.Connection;
  import java.sql.DatabaseMetaData;
  import java.sql.ResultSet;
  import java.sql.ResultSetMetaData;
  import java.sql.SQLException;
  import java.sql.Types;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.StringTokenizer;
  
  import javax.sql.DataSource;
  
  import org.javagen.agile.core.Generator;
  import org.javagen.agile.core.model.Model;
  import org.javagen.agile.db.model.Column;
  import org.javagen.agile.db.model.ColumnReference;
  import org.javagen.agile.db.model.Database;
  import org.javagen.agile.db.model.DbType;
  import org.javagen.agile.db.model.FkConstraint;
  import org.javagen.agile.db.model.FkEnum;
  import org.javagen.agile.db.model.PkColumn;
  import org.javagen.agile.db.model.Table;
  import org.javagen.agile.db.model.UniqueConstraint;
  import org.javagen.agile.db.util.DbUtil;
  
  /***
   * Create DB model tree from JDBC metadata from provided DataSource.
   * <p>
   * The scope of the model tree can be limited by specifying a list of 
   * <code>includeTableNames</code> and/or <code>excludeTableNames</code>, otherwise
   * all table schemas are loaded.
   *
   * @author Richard Easterling
   */
  public class DbLoader implements Generator {
  
      private static org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory.getLog(DbLoader.class);
  	private static final int CASCADE = DatabaseMetaData.importedKeyCascade; 
  	protected DataSource dataSource;
      protected List<String> includeTableNames;
      protected Set<String> excludeTableNames;
  	protected Model parentModel;
      protected boolean namesToUpperCase = false;
  
  	////////////////////////////////////////////////////////////////////////////
  	// Generator interface:
  	////////////////////////////////////////////////////////////////////////////
  	
  	/***
       * Create DB model tree from JDBC metadata from provided DataSource.
       * 
       * @param input can be one of three possibilities:
       * <ol>
       * <li>Database instance - provided database is processed instead of creating a new instance.</li>
       * <li>A non-Database instance - input model is treated as container for new Database instance.</li>
       * <li>null - new Database is created.</li>
       * </ol>
       * @return container model if present, otherwise return the Database instance.
  	 */
      public Model gen(Model input) {  
          Database database = null;
          Model result = null;
          if ( input!=null && (input instanceof Database) ) {
              //database instance provided, try to find a parent container to add it to
              database = (Database)input;
              if (parentModel!=null) {
                  parentModel.addChildModel(database);
                  result = parentModel;
              } else {
                  result = database;
              }
          } else {
              //create a new database instance and try to find a parent container to add it to
              database = new Database(Database.DEFAULT_MODEL_TYPE);
              if (parentModel!=null) {
                  parentModel.addChildModel(database);
                  result = parentModel;
              } else if (input!=null) {
                 input.addChildModel(database);
                 result = input;
             } else {
                 result = database;
             }
         }
         try {
         	load(database);
         } catch (SQLException e) {
         	throw new RuntimeException(e);
         }
         return result;
 	}
 
 	////////////////////////////////////////////////////////////////////////////
 	// Walk DB Schema and create Database tree:
 	////////////////////////////////////////////////////////////////////////////
 	
     /***
      * Load tables into Database instance using supplied DataSource.
      */
 	public void load(Database database) throws SQLException {
 		Connection dbconn = null;
 		List<String> tableList = includeTableNames==null ? null : includeTableNames;
 		try {
 			dbconn = dataSource.getConnection();
 			DatabaseMetaData dbMeta = dbconn.getMetaData();
 //			if (database.getName()==null)
 //				database.setName(dbMeta.getURL());
 			if (tableList==null) {
 				tableList = loadTableList(dbMeta, database);
             } else {
                 verifyTableList(dbMeta, database, tableList);
             }
 			loadTables(dbconn, dbMeta, database, tableList); //first pass loads tables/columns
 			loadImportedKeys(dbMeta, database, tableList); //second pass wires up foreign key relationships
 		} finally {
 			if (dbconn!=null) 
 				dbconn.close();			
 		}
 	}
 
     /*** 
      * Create list of all table names from DatabaseMetaData.
      */
     public List<String> loadTableList(DatabaseMetaData dbMeta, Database database) throws SQLException {
         String[] tableTypes = {"TABLE"}; //"TABLE", "VIEW", "SYSTEM TABLE", "GLOBAL TEMPORARY", "LOCAL TEMPORARY", "ALIAS", "SYNONYM"
         List<String> tableList = new ArrayList<String>();
         ResultSet tables = dbMeta.getTables(database.getCatalog(), database.getSchema(), null, tableTypes);
         while (tables.next()) {
             String tableName = tables.getString("TABLE_NAME");
             if (this.namesToUpperCase)
                 tableName = tableName.toUpperCase(); 
 //          String catalog = tables.getString("TABLE_CAT");
 //          String schema = tables.getString("TABLE_SCHEM");
 //          String tableType = tables.getString("TABLE_TYPE");
 //          String tableRemarks = tables.getString("REMARKS");
             if (excludeTableNames==null || !excludeTableNames.contains(tableName))
                 tableList.add(tableName);
         }
         tables.close();
         return tableList;
     }
     
     /*** 
      * Verify existence of provided table list.
      */
     public void verifyTableList(DatabaseMetaData dbMeta, Database database, List<String> tableList) throws SQLException {
         String[] tableTypes = {"TABLE"}; //"TABLE", "VIEW", "SYSTEM TABLE", "GLOBAL TEMPORARY", "LOCAL TEMPORARY", "ALIAS", "SYNONYM"
         for (String tableName : tableList) {
             ResultSet tables = dbMeta.getTables(database.getCatalog(), database.getSchema(), tableName, tableTypes);
             boolean OK = tables.next();
             tables.close();
             if (!OK) {
                 throw new IllegalArgumentException("Provided table name '"+tableName+"' not in database");
             }
         }
     }
     
 	/***
 	 * Populate column data for this table obtained from DatabaseMetaData.getColumns() ResultSet.
 	 * @todo add support for versioning by calling DatabaseMetaData.getVersionColumns()
 	 */
 	public void loadTables(Connection dbconn, DatabaseMetaData dbMeta, Database database, List<String> tableList) {
 		try {
 			for(String tableName : tableList) {
 				Table table = database.lookupTable(tableName); //table may already be present from serialized metadata
 				if (table==null) {
 					table = database.createTable();
 					table.setName(tableName);
 					database.addTable(table);
 				}
 				//preload primary key columns
 				Map<String, Short> pkMap = new HashMap<String, Short>();
 				ResultSet pkeys = dbMeta.getPrimaryKeys(null, database.getSchema(), tableName);
 				while (pkeys.next()) {
 					String pkColumnName = pkeys.getString("COLUMN_NAME");
 					Short seq = pkeys.getShort("KEY_SEQ");
 					pkMap.put(pkColumnName, seq);
 					Column col = table.lookupColumn(pkColumnName); //column may already be present from serialized metadata
 					PkColumn pkColumn = (PkColumn)col;
 					if (pkColumn==null) {
 						pkColumn = table.createPkColumn();
 						pkColumn.setName(pkColumnName);
 						table.addColumn(pkColumn);
 					}
 					pkColumn.setSequenceNumber(seq);
 				}
 				pkeys.close();
 
                 //load and set attributes of all columns
 				ResultSet columns = dbMeta.getColumns(database.getCatalog(), database.getSchema(), tableName, null);
 
                 int columnCount = 0;
 				StringBuffer logSB = log.isDebugEnabled() ? new StringBuffer(tableName+" table's columns: ") : null;
 				while (columns.next()) {
 					++columnCount;
 					String colName = columns.getString("COLUMN_NAME");
 					short colType = columns.getShort("DATA_TYPE");
 					int colSize = columns.getInt("COLUMN_SIZE");
 					String colTypeName = columns.getString("TYPE_NAME").toUpperCase();
 					if (logSB != null)
 						logSB.append(colName + ":" + colTypeName + ", ");
 					int scale = (colType == Types.DECIMAL || colType == Types.NUMERIC) ? columns.getInt("DECIMAL_DIGITS") : 0;
 					boolean nullable = (columns.getInt("NULLABLE") == DatabaseMetaData.columnNullable) ? true : false;
 					Object defaultValue = columns.getObject("COLUMN_DEF");
 					Column column = table.lookupColumn(colName); //column may already be present
 					if (column==null) {
 						column = table.createColumn();
 						column.setName(colName);
 						table.addColumn(column);
 					}
 					column.setColumnSize(colSize==0 ? null : colSize);
 					column.setDbType(DbType.lookup(colType));
 					column.setNotNull(nullable ? null : true);
 					column.setScale(scale==0 ? null : scale);
 					column.setDefaultValue(defaultValue);
 					column.setDbTypeName(colTypeName);
 				}
 				columns.close();
                 if (logSB!=null)
                     log.info(logSB.toString());
                 
                 //look for auto increment keys - is there a pure metadata way to do this?
                 ResultSetMetaData columnMetaData = dbconn.createStatement().executeQuery("select * from "+tableName+" where 1=0").getMetaData();
                 for(int i=1;i<=columnMetaData.getColumnCount();i++)
                     if (columnMetaData.isAutoIncrement(i)) {
                         log.info("auto increment column found:"+tableName+"."+columnMetaData.getColumnName(i));
                         Column autoColumn = table.lookupColumn(columnMetaData.getColumnName(i));
                         autoColumn.setAutoIncrement(true);
                     }
                 
                 //create unique constraints
                 Map<String,UniqueConstraint> uniqueConstraintsMap =  new HashMap<String,UniqueConstraint>();
                 ResultSet uniqueIndexes = dbMeta.getIndexInfo(database.getCatalog(), database.getSchema(), tableName, true, false);
                 while (uniqueIndexes.next()) {
                     String columnName = uniqueIndexes.getString("COLUMN_NAME");
                     Boolean nonUnique = uniqueIndexes.getBoolean("NON_UNIQUE");
                     String indexName = uniqueIndexes.getString("INDEX_NAME");
                     if (nonUnique!=null && !nonUnique) {
                         Column column = table.lookupColumn(columnName);
                         UniqueConstraint uniqueConstraint = uniqueConstraintsMap.get(indexName);
                         if (uniqueConstraint==null) {
                             uniqueConstraint = new UniqueConstraint();
                             uniqueConstraintsMap.put(indexName, uniqueConstraint);
                             uniqueConstraint.setName(indexName);
                         }
                         uniqueConstraint.addColumn(column);
                     }
                 }
                 uniqueIndexes.close();
                 //now add UniqueConstraint references to effected column contexts for easy reference
                 for(UniqueConstraint uniqueConstraint : uniqueConstraintsMap.values())
                     if (uniqueConstraint.columnReferencesSize()==1)
                         uniqueConstraint.getColumns().get(0).setUnique(true);
                 
                 
                 //look for idnentity columns
 //                columns = dbMeta.getColumns(database.getCatalog(), database.getSchema(), "SYSTEM_SEQUENCES", null);
 //                while (columns.next()) {
 //                    String colName = columns.getString("SEQUENCE_NAME");
 //                    System.out.println("SEQUENCE_NAME="+colName);
 //                }
 //                columns.close();
                 
 //              HSQL SYSTEM_SEQUENCES
 //              You can query the SYSTEM_SEQUENCES table for the next value that will be returned from any of 
 //              the defined sequences. The SEQUENCE_NAME column contains the name and the NEXT_VALUE 
 //              column contains the next value to be returned. 
             
 
                 
 			} // Table loop
 		} catch (SQLException e) {
 			throw new RuntimeException(e);
 		}
 	}
 
     /***
      * Wire up foreign key relationships 
      * @param dbMeta DatabaseMetaData instance
      * @param database root model node of DB tree
      * @param tableSet Mechinism for mapping subset of tables - FKs to tables not in this list are just treated like data columns. 
      * @throws java.sql.SQLException
      */
 	public void loadImportedKeys(DatabaseMetaData dbMeta, Database database, List<String> tableList) throws java.sql.SQLException
 	{
         Set<String> tableSet = new HashSet<String>(tableList);
 		for(String tableName : tableSet) {
 			ResultSet fkeys = dbMeta.getImportedKeys(null, database.getSchema(), tableName);
 			Table localTable = database.lookupTable(tableName);
 			StringBuffer logSB = log.isDebugEnabled() ? new StringBuffer(tableName+" table's imported foreign keys: ") : null;
 			while (fkeys.next()) {
 				String pkTableName    = fkeys.getString("PKTABLE_NAME");
 				String pkColName      = fkeys.getString("PKCOLUMN_NAME");
 				String fkTableName    = fkeys.getString("FKTABLE_NAME");
 				String fkColName      = fkeys.getString("FKCOLUMN_NAME");
 				short seq             = fkeys.getShort("KEY_SEQ");
                 String fkName         = fkeys.getString("FK_NAME");
                 boolean cascadeUpdate = CASCADE == fkeys.getInt("UPDATE_RULE");
                 boolean cascadeDelete = CASCADE == fkeys.getInt("DELETE_RULE");
 				boolean skipFk = tableSet!=null && !tableSet.contains(pkTableName);
 				if (logSB!=null) {
 					logSB.append(skipFk ? "SKIP>" : "FK").append(fkTableName).append(".").append(fkColName).append("[").append(seq).append("], ");
 				}
 				if (!skipFk) {
 					FkConstraint fkConstraint = localTable.lookupFkColumn(fkName);
 					Table foreignTable = database.lookupTable(pkTableName);
 					Column foreignColumn = foreignTable.lookupColumn(pkColName);
 					Column localColumn = localTable.lookupColumn(fkColName);
 					if (fkConstraint==null) {
 						fkConstraint = localTable.createFkColumn();
 						fkConstraint.setName(fkName);
                         localTable.addFkColumn(fkConstraint);
 					}
 					//fkColumn.setParentTable(localTable);
 					fkConstraint.setFkType(FkEnum.IMPORTED);
 					fkConstraint.setTargetTable(foreignTable);
                     if (cascadeUpdate) fkConstraint.setCascadeUpdate(cascadeUpdate);
                     if (cascadeDelete) fkConstraint.setCascadeDelete(cascadeDelete);
 					ColumnReference colRef = fkConstraint.lookupColumnReference(pkColName, fkColName);
 					if (colRef==null) {
 						colRef = fkConstraint.createColumnReference();
 						fkConstraint.addColumnReference(colRef);
 					}
 					colRef.setForeignColumn(foreignColumn);
 					colRef.setLocalColumn(localColumn);
 					colRef.setSequenceNumber(seq);
                     localColumn.put(ColumnReference.DEFAULT_MODEL_TYPE, colRef); //this identifies column as a FK
                     assert fkConstraint.getParentTable()!=null;
                     assert fkConstraint.getTargetTable()!=null;
                     
 				}
 			}
 			fkeys.close();
 			if (logSB!=null) {
 				log.info(logSB.toString());
 			}
             //create reverse (i.e. exported) relationships
             FkConstraint[] importedFkColumns = localTable.getFkColumns().toArray(new FkConstraint[localTable.getFkColumns().size()]);
             for(FkConstraint fkConstraint : importedFkColumns) {
                 try {
                     FkConstraint revFfColumn = DbUtil.createReverse(fkConstraint, true);
                     assert revFfColumn.getParentTable()!=null;
                     assert revFfColumn.getTargetTable()!=null;
                 } catch (Exception e) {
                     e.printStackTrace();
                 }
             }
             
             
             
 		} //tableName loop
         
 //		PKTABLE_CAT,PKTABLE_SCHEM, PKTABLE_NAME, PKCOLUMN_NAME, FKTABLE_CAT, FKTABLE_SCHEM, FKTABLE_NAME, FKCOLUMN_NAME, 	KEY_SEQ, UPDATE_RULE, DELETE_RULE, FK_NAME,   PK_NAME,               DEFERRABILITY, 
 //		null, 		PUBLIC, 	   PERSONS,      PERSON_ID, 	 null, 		 PUBLIC, 	    VEHICLES,    MODIFY_PERSON_ID, 	1, 	 	 3, 			3, 	   	   SYS_FK_98, SYS_REF_SYS_FK_98_105, 7, 
 //		null, 		PUBLIC, 	   PERSONS,      PERSON_ID, 	 null, 		 PUBLIC, 	    VEHICLES,    CREATE_PERSON_ID, 	1, 	 	 3, 			3,  	   SYS_FK_97, SYS_REF_SYS_FK_97_103, 7, 
 //		null, 		PUBLIC, 	   ZIPCODES,     ZIPCODE, 	     null, 		 PUBLIC, 	    VEHICLES,  REGISTRATION_ZIPCODE,1, 	 	 3, 			3, 	       SYS_FK_96, SYS_REF_SYS_FK_96_101, 7, 
 	}
 
     ////////////////////////////////////////////////////////////////////////////
     // Injected getters and setters:
     ////////////////////////////////////////////////////////////////////////////
     
     public List<String> getIncludeTableNames() {
         return includeTableNames;
     }
 
     public void setIncludeTableNames(List<String> tableNames) {
         if (namesToUpperCase && tableNames!=null) {
             for(int i=0;i<tableNames.size();i++)
                 tableNames.set(i, tableNames.get(i).toUpperCase());
         }
         this.includeTableNames = tableNames;
     }
 
     public DataSource getDataSource() {
         return dataSource;
     }
 
     public void setDataSource(DataSource dataSource) {
         this.dataSource = dataSource;
     }
 
     public Model getParentModel() {
         return parentModel;
     }
 
     public void setParentModel(Model baseModel) {
         this.parentModel = baseModel;
     }
 
     public boolean isNamesToUpperCase() {
         return namesToUpperCase;
     }
 
     public void setNamesToUpperCase(boolean namesToUpperCase) {
         this.namesToUpperCase = namesToUpperCase;
     }
 
     public Set<String> getExcludeTableNames() {
         return excludeTableNames;
     }
 
     public void setExcludeTableNames(Set<String> excludeTableNames) {
         if (excludeTableNames==null) {
             this.excludeTableNames = null;
         } else {
             this.excludeTableNames = new HashSet<String>();
             for(String name : excludeTableNames)
                 this.excludeTableNames.add( this.namesToUpperCase ? name.toUpperCase() : name );
         }
     }
 
     /***
      * White-space/comma delineated list of table names to include in the transformation process.
      */
     public void setIncludeTables(String delineatedIncludeTables) {
         if (delineatedIncludeTables==null || delineatedIncludeTables.trim().length()==0)
             return;
         List<String> included = new ArrayList<String>();
         StringTokenizer toks = new StringTokenizer(delineatedIncludeTables,", ", false);
         while(toks.hasMoreTokens())
             included.add( toks.nextToken() );
         setIncludeTableNames(included);
     }
 
     /***
      * White-space/comma delineated list of table names to exclude from the transformation process.
      */
     public void setExcludeTables(String delineatedExcludeTables) {
         if (delineatedExcludeTables==null || delineatedExcludeTables.trim().length()==0)
             return;
         Set<String> excluded = new HashSet<String>();
         StringTokenizer toks = new StringTokenizer(delineatedExcludeTables,", ", false);
         while(toks.hasMoreTokens())
             excluded.add( toks.nextToken() );
         setExcludeTableNames(excluded);
     }
 
 }