/*
 * This file is part of the DistSim distributed simulation framework (wrapper)
 * Copyright (C) 2007 Ulf Hermann
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */
package brn.distsim.wrapper;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.rmi.RemoteException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.Set;

public class DBConnector {
	private Connection serializedConnection;

	private Connection relaxedConnection;

	private Connection resultsConnection;

	private PreparedStatement getSimulations;

	private PreparedStatement assignSimulation;

	private PreparedStatement writeFile;

	private PreparedStatement writeStdOut;

	private PreparedStatement updateHosts;

	private int numStdOutChunks;

	private int groupId;

	private List inFiles;

	private List packages;

	private int currentSimulationId = 0;

	private Set blackList;

	private PreparedStatement writeStdErr;

	private Main.ConfigData configuration;

  private String workingDir;

	public DBConnector(Main.ConfigData config, String workingDir) throws SQLException {
		this.configuration = config;
		inFiles = new LinkedList();
		packages = new LinkedList();
		blackList = new HashSet();
		this.workingDir = workingDir;
		initConnections();
	}

  /**
   * @return the blackList
   */
  public Set getBlackList() {
    return blackList;
  }

	void initConnections() throws SQLException {
	  close();

		try {
			Class.forName("com.mysql.jdbc.Driver");
			String dbUrl = "jdbc:mysql://" + configuration.getDefinitionsHost()
					+ "/" + configuration.getDefinitionsDatabase();
			String dbUser = configuration.getDefinitionsUsername();
			String dbPassword = configuration.getDefinitionsPassword();
			serializedConnection = DriverManager.getConnection(dbUrl, dbUser,
					dbPassword);
			serializedConnection.createStatement().execute(
					"SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;");
			serializedConnection.createStatement().execute("SET AUTOCOMMIT=0;");
			relaxedConnection = DriverManager.getConnection(dbUrl, dbUser,
					dbPassword);
			getSimulations = serializedConnection
					.prepareStatement("SELECT * FROM simulations WHERE assigned<=>NULL FOR UPDATE;");
			assignSimulation = serializedConnection
					.prepareStatement("UPDATE simulations SET assigned = "
							+ configuration.getHostId() + " WHERE id = ?;");
			String resUrl = "jdbc:mysql://" + configuration.getResultsHost()
					+ "/" + configuration.getResultsDatabase();
			resultsConnection = DriverManager.getConnection(resUrl,
					configuration.getResultsUsername(), configuration
							.getResultsPassword());
			writeFile = resultsConnection
					.prepareStatement("INSERT IGNORE INTO out_files VALUES (?, ?, ?);");
			writeStdOut = resultsConnection
					.prepareStatement("INSERT INTO stdout VALUES (?, ?, ?);");
			writeStdErr = resultsConnection
					.prepareStatement("INSERT INTO stderr VALUES (?, ?, ?);");

			String description = configuration.getHostDescription();
			updateHosts = relaxedConnection
					.prepareStatement("INSERT INTO hosts VALUES('"
							+ configuration.getHostId() + "', '"
							+ configuration.getHostName() + "', '"
							+ configuration.getHostArchitecture() + "', '"
							+ description
							+ "') ON DUPLICATE KEY UPDATE description='"
							+ description + "';");
		} catch (ClassNotFoundException e) {
			// TODO Automatisch erstellter Catch-Block
			e.printStackTrace();
		}
	}

	public boolean readFilesFromDB() {
	  ResultSet files = null;
		try {
			files = relaxedConnection.createStatement().executeQuery(
					"SELECT * FROM group_in_files WHERE groupId = '" + groupId
							+ "';");
			while (files.next()) {
				String path = files.getString("remotePath");
				ResultSet file = relaxedConnection.createStatement()
						.executeQuery(
								"SELECT * FROM in_files WHERE id = '"
										+ files.getInt("fileId") + "';");
				file.next();
				FileOutputStream writer = null;
				InputStream stream = null;
				try {
	        writer = new FileOutputStream(workingDir + path, false);
	        stream = file.getBinaryStream("content");
	        int read = 0;
	        byte[] data = new byte[1024];
	        while ((read = stream.read(data)) > 0)
	          writer.write(data, 0, read);
				} finally {
				  if (null != writer)
				    writer.close();
				  if (null != stream)
				    stream.close();
				}
				inFiles.add(path);
			}
		} catch (SQLException e) {
			// TODO Automatisch erstellter Catch-Block
			e.printStackTrace();
			return false;
		} catch (IOException e) {
			// TODO Automatisch erstellter Catch-Block
			e.printStackTrace();
      return false;
		}
		return true;
	}

	private void deleteResultFiles() {
		try {
			ResultSet files;
			files = relaxedConnection.createStatement().executeQuery(
					"SELECT * FROM group_out_files WHERE groupId = '" + groupId
							+ "';");
			while (files.next()) {
				String path = files.getString("remotePath");
				File file = new File(workingDir + path);
				if (!file.exists())
					continue;
				if (file.isDirectory())
					deleteDirectory(file);
				else {
					file.delete();
				}
			}
		} catch (SQLException e) {
			// TODO Automatisch erstellter Catch-Block
			e.printStackTrace();
		}
	}

	private void deleteDirectory(File dir) {
		File[] files = dir.listFiles();
		for (int i = 0; i < files.length; ++i) {
			File file = files[i];
			if (file.isDirectory())
				deleteDirectory(file);
			else {
				file.delete();
			}
		}
		dir.delete();
	}

	private void writeFilesToDB() {
		try {
			ResultSet files;
			files = relaxedConnection.createStatement().executeQuery(
					"SELECT * FROM group_out_files WHERE groupId = '" + groupId
							+ "';");
			while (files.next()) {
				String path = files.getString("remotePath");
				File file = new File(workingDir + path);
				if (file.isDirectory())
					writeDirectoryToDB(file);
				else {
					writeFileToDB(file);
				}
			}
		} catch (SQLException e) {
			// TODO Automatisch erstellter Catch-Block
			e.printStackTrace();
		}
	}

	private void writeFileToDB(File file) {
    String path = file.getPath();
		try {
			writeFile.setInt(1, currentSimulationId);
			writeFile.setString(2, path);
			// unfortunately we can't save files larger than 4GB, but who cares
			// ...
			writeFile.setBinaryStream(3, new FileInputStream(file), (int) file
					.length());
			writeFile.executeUpdate();
			file.delete();
		} catch (SQLException e) {
			// TODO Automatisch erstellter Catch-Block
			e.printStackTrace();
		} catch (FileNotFoundException e) {
		  System.err.println("Expected file ("+path
		      +") des not exist, not stored in database!");
		}
	}

	public void writeStdOutToDB(InputStream out) throws InterruptedIOException {
		writeChunksToDB(out, writeStdOut);
	}

	public void writeStdErrToDB(InputStream err) throws InterruptedIOException {
		writeChunksToDB(err, writeStdErr);
	}

	private void writeChunksToDB(InputStream in, PreparedStatement sttmt) throws InterruptedIOException {
		try {
			sttmt.setInt(1, currentSimulationId);
			byte out[] = new byte[1024];
			int read = 0;
			while ((read = in.read(out)) > 0) {
				sttmt.setInt(2, numStdOutChunks++);
				if (read < 1024) {
					// the last chunk isn't aligned by 1024
					byte truncated[] = new byte[read];
					System.arraycopy(out, 0, truncated, 0, read);
					sttmt.setBytes(3, truncated);
				} else
					sttmt.setBytes(3, out);
				// TODO throws an exception if the connection timed out!!
				sttmt.executeUpdate();
			}
		} catch (SQLException e) {
			// TODO Automatisch erstellter Catch-Block
			e.printStackTrace();
		} catch (InterruptedIOException e) {
		  throw e;
		} catch (IOException e) {
			// TODO Automatisch erstellter Catch-Block
			e.printStackTrace();
		}
	}

	private void writeDirectoryToDB(File dir) {
		File[] files = dir.listFiles();
		for (int i = 0; i < files.length; ++i) {
			File file = files[i];
			if (file.isDirectory())
				writeDirectoryToDB(file);
			else {
				writeFileToDB(file);
			}
		}
	}

	public boolean readPackages() {
		try {
			int numPkgs = 0;
			ResultSet studyPkgs = relaxedConnection
					.createStatement()
					.executeQuery(
							"SELECT * FROM study_packages WHERE studyId=(SELECT studyId FROM groups WHERE id = "
									+ groupId + ");");
			while (studyPkgs.next())
				numPkgs++;

			ResultSet pkgs = relaxedConnection
					.createStatement()
					.executeQuery(
							"SELECT DISTINCT url, localPath FROM packages INNER JOIN study_packages ON "
									+ "(packages.name = study_packages.packageName "
									+ "AND packages.version=study_packages.packageVersion) "
									+ "WHERE (architecture='all' OR architecture='"
									+ configuration.getHostArchitecture()
									+ "') "
									+ "AND studyId=(SELECT studyId FROM groups WHERE id = "
									+ groupId + ");");
			while (pkgs.next()) {
				Package pkg = new Package(pkgs.getString("url"),
				    workingDir + pkgs.getString("localPath"));
				numPkgs--;
				packages.add(pkg);
			}
			if (numPkgs != 0) {
				// there have been package descriptions without instances for
				// our architecture
				// we need to reject this simulation
				packages.clear();
				return false;
			} else {
				Iterator i = packages.iterator();
				while (i.hasNext()) {
					Package pkg = (Package) i.next();
					pkg.download();
					pkg.unzip();
				}
				return true;
			}
		} catch (SQLException e) {
			e.printStackTrace();
			return false;
		} catch (IOException e) {
      e.printStackTrace();
      return false;
    }
	}

  public Simulation query(Properties p) throws SQLException {
    checkConnection(serializedConnection);
    checkConnection(relaxedConnection);
    checkConnection(resultsConnection);
    ResultSet newSimulations = null;
    try {
      updateHosts.execute();
      serializedConnection.createStatement()
          .execute("START TRANSACTION;");

      newSimulations = getSimulations.executeQuery();

      do {
        if (!newSimulations.next()) {
          serializedConnection.createStatement().execute("COMMIT;");
          return null;
        } else {
          currentSimulationId = newSimulations.getInt("id");
        }
      } while (blackList.contains(new Integer(currentSimulationId)));

      assignSimulation.setInt(1, currentSimulationId);
      assignSimulation.execute();
      serializedConnection.createStatement().execute("COMMIT;");

    } catch (SQLException e) {
      try {
        serializedConnection.createStatement().execute("ROLLBACK;");
      } catch (SQLException e1) {
        e1.printStackTrace();
        throw e1;
      }
      // TODO Automatisch erstellter Catch-Block
      e.printStackTrace();
      // throw to re-init the db connection
      throw e;
    }
    try {

      groupId = newSimulations.getInt("groupId");
      ResultSet group = relaxedConnection.createStatement().executeQuery(
          "SELECT studyId FROM groups WHERE id = '" + groupId + "';");
      group.next();
      int studyId = group.getInt("studyId");
      ResultSet template = relaxedConnection.createStatement()
          .executeQuery(
              "SELECT * FROM studies WHERE id = '" + studyId
                  + "';");

      numStdOutChunks = 0;
      ResultSet params = relaxedConnection.createStatement()
          .executeQuery(
              "SELECT * FROM params WHERE simulationId = '"
                  + currentSimulationId + "';");

      resultsConnection.createStatement().execute(
          "INSERT IGNORE INTO progress VALUES ('"
              + currentSimulationId
              + "', 'RUNNING', NOW(), NULL, '0');");
      return new Simulation(template, params, p, currentSimulationId,
          this);
    } catch (SQLException e) {
      try {
        // here we know the relaxedConnection should work OK, as the
        // previous try-catch
        // block was passed (updateHosts ...)
        relaxedConnection.createStatement().execute(
            "UPDATE simulations SET assigned = NULL WHERE id = "
                + currentSimulationId + ";");
      } catch (SQLException e1) {
        e1.printStackTrace();
        throw e1;
      }
      // TODO Automatisch erstellter Catch-Block
      e.printStackTrace();
      // throw to re-init the db connection
      throw e;
    } catch (RemoteException e) {
      // TODO Automatisch erstellter Catch-Block
      e.printStackTrace();
    }
    return null;
  }
  
  public boolean queryPackagesAndFiles(Simulation simulation) throws SQLException {
    checkConnection(serializedConnection);
    checkConnection(relaxedConnection);
    checkConnection(resultsConnection);
      if (!readPackages()) {
//        int oldId = currentSimulationId;
//        blackList.add(new Integer(oldId));
//        System.out.println("rejected simulation " + oldId
//            + ": undefined packages");
        return false;
      }

      if (!readFilesFromDB()) {
//        int oldId = currentSimulationId;
//        blackList.add(new Integer(oldId));
//        System.out.println("rejected simulation " + oldId
//            + ": undefined packages");
        return false;
      }

    return true;
  }

	public void jobFinished() throws SQLException {
		checkConnection(resultsConnection);
		resultsConnection
				.createStatement()
				.execute(
						"UPDATE progress SET state='FINISHED', expectedEnd=NOW(), percentageDone='100' WHERE simulationId="
								+ currentSimulationId + ";");
		writeFilesToDB();
		cleanup();
	}

	private void checkConnection(Connection c) throws SQLException {
		try {
			// dummy statement to trigger an exception if connection has timed
			// out
			Statement stmt = c.createStatement();
      stmt.execute("SHOW STATUS;");
      while (!((stmt.getMoreResults() == false) && (stmt.getUpdateCount() == -1))) ;
      stmt.close();
		} catch (SQLException e) {
			initConnections();
		}
	}

	public void jobCanceled() throws SQLException {
		checkConnection(resultsConnection);
		resultsConnection.createStatement().execute(
				"UPDATE progress SET state='CANCELED', expectedEnd=NULL WHERE simulationId="
						+ currentSimulationId + ";");
		writeFilesToDB();

		cleanup();
	}

	public void jobRepeated() throws SQLException {
		checkConnection(resultsConnection);
		resultsConnection
				.createStatement()
				.execute(
						"UPDATE progress SET started=NOW(), percentageDone='0', expectedEnd=NULL WHERE simulationId="
								+ currentSimulationId + ";");
		deleteStdOutFromDB();
		deleteStdErrFromDB();
		cleanup();
		deleteResultFiles();

		checkConnection(relaxedConnection);
		readFilesFromDB();
		readPackages();
	}

	private void cleanup() {
		Iterator i = inFiles.iterator();
		while (i.hasNext()) {
			new File((String) i.next()).delete();
		}
		inFiles.clear();

		i = packages.iterator();
		while (i.hasNext()) {
			((Package) i.next()).delete();
		}
		packages.clear();
	}

	private void deleteStdOutFromDB() throws SQLException {
		resultsConnection.createStatement().execute(
				"DELETE FROM stdout WHERE simulationId=" + currentSimulationId
						+ ";");
	}

	private void deleteStdErrFromDB() throws SQLException {
		resultsConnection.createStatement().execute(
				"DELETE FROM stderr WHERE simulationId=" + currentSimulationId
						+ ";");
	}

	public void close() {
	  try {
      if (null != updateHosts)
        updateHosts.close();
    } catch (Throwable e) {
    }
    try {
      if (null != writeStdErr)
        writeStdErr.close();
    } catch (Throwable e) {
    }
    try {
      if (null != writeStdOut)
        writeStdOut.close();
    } catch (Throwable e) {
    }
    try {
      if (null != writeFile)
        writeFile.close();
    } catch (Throwable e) {
    }
    try {
      if (null != resultsConnection)
        resultsConnection.close();
    } catch (Throwable e) {
    }
    try {
      if (null != assignSimulation)
        assignSimulation.close();
    } catch (Throwable e) {
    }
    try {
      if (null != getSimulations)
        getSimulations.close();
    } catch (Throwable e) {
    }
    try {
      if (null != relaxedConnection)
        relaxedConnection.close();
    } catch (Throwable e) {
    }
    try {
      if (null != serializedConnection)
        serializedConnection.close();
    } catch (Throwable e) {
    }

	  updateHosts = null;
    writeStdErr = null;
    writeStdOut = null;
    writeFile = null;
    resultsConnection = null;
    assignSimulation = null;
    getSimulations = null;
    relaxedConnection = null;
    serializedConnection = null;
	}

}
