/*
 * $Source: /opt/cvsroot/erserver/src/com/postgres/replic/util/SlaveProcessor.java,v $
 * $Author: ronz $ $Revision: 1.4 $ $Date: 2003/12/15 10:26:37 $
 *
 */

package com.postgres.replic.util;

import com.postgres.replic.server.props.ServerProps;
import com.postgres.replic.util.struct.CommandType;
import com.postgres.replic.util.struct.LastSyncData;
import com.postgres.replic.util.struct.SnapshotData;
import com.postgres.replic.util.struct.SnapshotDataBlock;
import com.postgres.replic.util.struct.TableMap;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;

public class SlaveProcessor extends DBProcessor {

	private TableMap tableMap; // of TableMap objects
	//private int server = 0;
	private LastSyncData lastSync = new LastSyncData();
	//private boolean gotSYNCID = false;
	//private boolean tableMapSet = false;
	private static final int ID_NULL = -999;
	protected static final int DUPLICATE_ROW = -9999;
	private long lastsid = ID_NULL;
	private boolean resetTableMap = false;

	public SlaveProcessor reset(Connection conn, int server, ServerProps serverProps)
			throws RservException, Exception {
		super.init(conn, serverProps);
		lastSync.reset();

		//this.server = server;
	//	gotSYNCID = false;
	//	tableMapSet = false;

		if (resetTableMap) {
			getLogger().info("SlaveProcessor::reset: Resetting table map");
			tableMap = null;
			tableMap = getTableMap();
			resetTableMap = false;
		}

		setTableColumnCache();
		resetPstmtCache();
		setInitialized(true);

		return this;
	}

	/* Returns last SyncID applied on Slave
	*/

	protected long GetSyncID() throws RservException {
		validateInit();
		ResultSet rs = null;
		Statement stmt = null;
		String sql = "select max(syncid) from "+serverProps.createRelName("_RSERV_SLAVE_SYNC_");
		getLogger().debug("SlaveProcessor::GetSyncID: sql=" + sql);
		long rc = T_NO;
		try {
			stmt = getConnection().createStatement();
			rs = stmt.executeQuery(sql);
			while (rs.next()) {
				rc = rs.getLong(1);
			}
			if (getDebug()) getLogger().debug("SlaveProcessor::GetSyncID: max syncid in _RSERV_SLAVE_SYNC_=" + rc);
		} catch (Exception e) {
			throw new RservException(
					"SlaveProcessor::GetSyncID: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}

		}
		return rc;
	}

	protected void setConstraintsDeferred() throws RservException {
		validateInit();
		Statement stmt = null;
		String sql = "SET CONSTRAINTS ALL DEFERRED";
		getLogger().debug("SlaveProcessor::setConstraintsDeferred: sql=" + sql);
		try {
			stmt = getConnection().createStatement();
			stmt.execute(sql);
		} catch (Exception e) {
			throw new RservException(
					"SlaveProcessor::setConstraintsDeferred: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}
	}

	/* MAP oid --> tabname, keyname
	 * retrieve it once since table struct does not change
	*/

	protected TableMap getTableMap() throws RservException {
		if (tableMap != null) {
			if (getDebug())
				getLogger().debug("SlaveProcessor::getTableMap: tableMap is set tableMap=" +
						tableMap);
			return tableMap;
		}

		ResultSet rs = null;
		Statement stmt = null;
		String sql = null;

		sql = "select pgc.oid, pgc.relname, pgc.attname, rt.key" +
				" from "+serverProps.createRelName("_RSERV_SLAVE_TABLES_")+" rt, "+
				serverProps.createRelName("_ers_class_attr")+" pgc" +
				" where pgc.oid = rt.reloid and pgc.attnum = rt.key";

		try {
			tableMap = new TableMap();
			stmt = getConnection().createStatement();
			getLogger().debug("SlaveProcessor::getTableMap: tableMap is set sql=" + sql);
			rs = stmt.executeQuery(sql);
			while (rs.next()) {
				String oid = rs.getString(1);
				String relName = rs.getString(2);
				String attName = rs.getString(3);
				String key = rs.getString(4);

				// Adding element with row key = relName;
				tableMap.addElement(relName).
						setOid(oid).
						setRelName(relName).
						setAttName(attName).
						setKey(key);
			}
			//tableMapSet = true;
		} catch (Exception e) {
			getLogger().error("SlaveProcessor::getTableMap: ", e);
			throw new RservException(
					"SlaveProcessor::getTableMap-1: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}

		return tableMap;
	}

	//private long getLastSid() {
	//	return lastsid;
	//}

	private void setLastSid(long lastsid) {
		this.lastsid = lastsid;
	}


	/*  If _RSERV_SLAVE_SYNC_ is empty then - return 0 - correspond to 'A' any
		snapshot in Rserv.pm
	*/

	private int doSyncId(SnapshotDataBlock snapDataBlock)
			throws RservException {
		ResultSet rs = null;
		Statement stmt = null;
		int row = 0;
		long syncid_old = 0;

		String sql = "select syncid, synctime from "
				+ serverProps.createRelName("_RSERV_SLAVE_SYNC_")+" where syncid = "
				+ "(select max(syncid) from "+serverProps.createRelName("_RSERV_SLAVE_SYNC_")+")";

		getLogger().debug("SlaveProcessor::doSyncId: sql(1)=" + sql);
		if (getDebug()) getLogger().debug("SlaveProcessor::doSyncId: snapDataBlock=" + snapDataBlock);
		try {
			if (snapDataBlock.getCommandInt() != CommandType.SYNCID) {
				return T_NO;
			}

			long syncid = snapDataBlock.getParamLong();

			stmt = getConnection().createStatement();
			rs = stmt.executeQuery(sql);

			while (rs.next()) {
				row++;
				syncid_old = rs.getLong(1);
			}
			rs.close();

			if (row == 0) {
				sql = "insert into "
						+serverProps.createRelName("_RSERV_SLAVE_SYNC_")+"(syncid, synctime) values ("
						+ syncid + ", now())";
				getLogger().debug("SlaveProcessor::doSyncId: sql(2)=" + sql);
				stmt.executeUpdate(sql);
			} else if (syncid_old >= syncid) {
				if (getDebug())
					getLogger().debug("SlaveProcessor::doSyncId: syncid_old >= syncid; syncid_old="
							+ syncid_old);
			} else {
				sql = "update "+serverProps.createRelName("_RSERV_SLAVE_SYNC_") +
						" set syncid = " + syncid + ", synctime = now()";
				getLogger().debug("SlaveProcessor::doSyncId: sql(3)=" + sql);
				stmt.executeUpdate(sql);
			}

		} catch (Exception e) {
			getLogger().debug("SlaveProcessor::doSyncId: ", e);
			throw new RservException(
					"SlaveProcessor::doSyncId: " + e.toString());
		} finally {
			try {
				if (stmt != null) {
					stmt.close();
				}
			} catch (Exception ex) {
			}
		}
		setLastSid(syncid_old);
		return row;
	}


	/*
	 */

	private int doDelete(SnapshotDataBlock snapDataBlock)
			throws RservException {

		getLogger().debug("SlaveProcessor::doDelete: BEGIN");
		PreparedStatement pstmt = null;
		int rc = 0;

		try {
			if (snapDataBlock.getCommandInt() != CommandType.DELETE) {
				return 0;
			}
			TableMap tableMap = getTableMap();
			String tabname = snapDataBlock.getParam();

			// NEW for ASYMMETRIC slave update:
			if (!tableMap.rowExists(tabname)) {
				getLogger().debug("SlaveProcessor::DoDelete: table " +
						tabname + " does not exists - return 0");
				return 0;
			}

			pstmt = getTableColumnCache().getPstmtDelete(tabname);
			getLogger().debug("SlaveProcessor::doDelete: sql=" + getTableColumnCache().getSqlDelete(tabname));
			for (int i = 0; i < snapDataBlock.size(); i++) {
				String key = snapDataBlock.getRow(i).getKey();
				// Bind:
				if (key != null) {
					pstmt.setString(1, key);
				} else {
					pstmt.setNull(1, java.sql.Types.NULL);
				}

				if (getDebug()) getLogger().debug("SlaveProcessor::doDelete: key=" + key);
				rc += pstmt.executeUpdate();
			}
			getLogger().debug("SlaveProcessor::doDelete: DELETED " + rc + " ROWS");

		} catch (Exception e) {
			throw new RservException(
					"SlaveProcessor::doDelete: " + e.toString());
		}
		return rc;
	}


	/*
	*/

	private int doUpdate(SnapshotDataBlock snapDataBlock)
			throws RservException {

		int rc = 0;
		getLogger().debug("SlaveProcessor::doUpdate: BEGIN");

		try {
			if (snapDataBlock.getCommandInt() != CommandType.UPDATE &&
					snapDataBlock.getCommandInt() != CommandType.INSERT) {
				return rc;
			}

			String tabname = snapDataBlock.getParam();
			// NEW for ASYMMETRIC slave update:
			if (!tableMap.rowExists(tabname)) {
				getLogger().debug("SlaveProcessor::DoUpdate: table " +
						tabname + " does not exists - return 0");
				return 0;
			}

			for (int i = 0; i < snapDataBlock.size(); i++) {
				int rc1 = updateRow(tabname, snapDataBlock, i);
				if (getDebug()) getLogger().debug("SlaveProcessor::DoUpdate: table rc1=" + rc1 + "; snapDataBlock.size()=" + snapDataBlock.size());
				if (rc1 <= 0) {
					getLogger().debug("SlaveProcessor::DoUpdate: insertRow INSIDE OF doUpdate" + rc1);
					rc += insertRow(tabname, snapDataBlock, i);
				} else {
					rc += rc1;
				}
			}
		} catch (Exception e) {
			getLogger().error("SlaveProcessor::DoUpdate: ", e);
			throw new RservException("SlaveProcessor::DoUpdate: " + e.toString());
		} finally {

		}

		return rc;
	}

	/*
	*/

	private int updateRow(String tabname, SnapshotDataBlock snapDataBlock, int idx)
			throws RservException {
		getLogger().debug("SlaveProcessor::updateRow: BEGIN; tabname=" + tabname + "; idx=" + idx);
		PreparedStatement pstmt = null; // changed because some strings cannot be loded otherwise
		int rc = T_NO;
		TableMap tableMap = getTableMap();
		String keyValue = null;

		try {
			pstmt = getTableColumnCache().getPstmtUpdate(tabname);
			int oidkey = snapDataBlock.getRow(idx).getKeyInt();
			boolean useoid = snapDataBlock.getRow(idx).useOid();
			if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: ; oidkey=" + oidkey);
			long oidval = oidkey;
			if (useoid) {
				oidval = snapDataBlock.getRow(idx).getLong(0);
				if (oidval <= 0) {
					throw new RservException("SlaveProcessor::updateRow: Invalid OID oidval=" + oidval + "; idx=" + idx + "; oidkey=" + oidkey + "; snapDataBlock.getRow(idx).size()=" + snapDataBlock.getRow(idx).size());
				}
			}

			if (useoid) {
				keyValue = "" + oidval;
				if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: oidval > 0: oidval=" + oidval);
			} else {  // normal - our case
				int key = tableMap.setRow(tabname).getKeyInt() - 1;
				if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: oidval <= 0 key=" + key);
				if (key < 0) {
					throw new RservException(
							"SlaveProcessor::updateRow: key < 0!");
				}
				keyValue = snapDataBlock.getRow(idx).get(key);
			}

			// Bind params:
			getLogger().debug("SlaveProcessor::updateRow: sql(4)=" + getTableColumnCache().getSqlUpdate(tabname));

			int count = 0;
			for (int i = 0; i < getTableColumnCache().size(tabname); i++) {
				if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: sql(4-5-4) BEGIN i=" + i);
				String val = snapDataBlock.getRow(idx).get(i);
				if (val != null) {
					if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: val != null sql(bet 4-5), i=" + i + "; val=" + val);
					pstmt.setString(i + 1, val);
					if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: val != null sql(5), i=" + i + "; val=" + val);
				} else {
					if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: val == null sql(bet 4-6), i=" + i + "; val=" + val);
					pstmt.setNull(i + 1, java.sql.Types.NULL);
				}
				count++;
				if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: sql(4-5-4) END i=" + i);
			}
			count++;
			if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: before Bind key column:  keyValue=" + keyValue);

			// Bind key column:
			if (keyValue != null) {
				pstmt.setString(count, keyValue);
			} else {
				pstmt.setNull(count, java.sql.Types.NULL);
			}

			if (getDebug()) getLogger().debug("SlaveProcessor::updateRow: sql(6), before pstmt.executeUpdate()");
			rc = pstmt.executeUpdate();
			getLogger().debug("SlaveProcessor::updateRow: UPDATED " + rc + " ROWS");
		} catch (Exception e) {
			e.printStackTrace();
			throw new RservException("SlaveProcessor::updateRow: " + e.toString());
		}
		return rc;
	}

	private int deleteDuplicateRow(String tabname, SnapshotDataBlock snapDataBlock, int idx)
			throws RservException {
		getLogger().debug("SlaveProcessor::deleteDuplicateRow: BEGIN; tabname=" + tabname + "; idx=" + idx);
		PreparedStatement pstmt = null; // changed because some strings cannot be loded otherwise
		int rc = T_NO;

		String keyValue = null;

		try {
			pstmt = getTableColumnCache().getPstmtDelete(tabname);
			int oidkey = snapDataBlock.getRow(idx).getKeyInt();
			boolean useoid = snapDataBlock.getRow(idx).useOid();
			if (getDebug()) getLogger().debug("SlaveProcessor::deleteDuplicateRow: ; oidkey=" + oidkey);
			long oidval = oidkey;
			if (useoid) {
				oidval = snapDataBlock.getRow(idx).getLong(0);
				if (oidval <= 0) {
					throw new RservException("SlaveProcessor::deleteDuplicateRow: Invalid OID oidval=" + oidval + "; idx=" + idx + "; oidkey=" + oidkey + "; snapDataBlock.getRow(idx).size()=" + snapDataBlock.getRow(idx).size());
				}
			}

			if (useoid) {
				keyValue = "" + oidval;
				if (getDebug()) getLogger().debug("SlaveProcessor::deleteDuplicateRow: oidval > 0: oidval=" + oidval);
			} else {  // normal - our case
				int key = tableMap.setRow(tabname).getKeyInt() - 1;
				if (getDebug()) getLogger().debug("SlaveProcessor::deleteDuplicateRow: oidval <= 0 key=" + key);
				if (key < 0) {
					throw new RservException(
							"SlaveProcessor::deleteDuplicateRow: key < 0!");
				}
				keyValue = snapDataBlock.getRow(idx).get(key);
			}

			// Bind params:
			getLogger().info("DELETING DuplicateRow: sql=" + getTableColumnCache().getSqlDelete(tabname));
			getLogger().info("DELETING DuplicateRow: keyValue=" + keyValue);

			// Bind key column:
			if (keyValue != null) {
				pstmt.setString(1, keyValue);
			}

			if (getDebug()) getLogger().debug("SlaveProcessor::deleteDuplicateRow: sql(6), before pstmt.executeUpdate()");
			rc = pstmt.executeUpdate();
			getLogger().debug("SlaveProcessor::deleteDuplicateRow: DELETED " + rc + " ROWS");
		} catch (Exception e) {
			throw new RservException("SlaveProcessor::deleteDuplicateRow: " + e.toString());
		} finally {

		}
		return rc;
	}

	/*
	*/

	private int insertRow(String tabname, SnapshotDataBlock snapDataBlock, int idx)
			throws RservException {
		getLogger().debug("SlaveProcessor::insertRow: BEGIN");
		PreparedStatement pstmt = null; // changed because some strings cannot be loded otherwise
		int rc = T_NO;
		//TableMap tableMap = getTableMap();
		long time = System.currentTimeMillis();

		try {
			pstmt = getTableColumnCache().getPstmtInsert(tabname);
			// Bind params:
			getLogger().debug("SlaveProcessor::insertRow: sql(4)=" + getTableColumnCache().getSqlInsert(tabname));

			for (int i = 0; i < getTableColumnCache().size(tabname); i++) {
				if (getDebug()) getLogger().debug("SlaveProcessor::insertRow: sql(4-5-4) BEGIN i=" + i);
				String val = snapDataBlock.getRow(idx).get(i);
				if (val != null) {
					if (getDebug()) getLogger().debug("SlaveProcessor::insertRow: val != null sql(bet 4-5), i=" + i + "; val=" + val);
					pstmt.setString(i + 1, val);
					if (getDebug()) getLogger().debug("SlaveProcessor::insertRow: val != null sql(5), i=" + i + "; val=" + val);
				} else {
					if (getDebug()) getLogger().debug("SlaveProcessor::insertRow: val == null sql(bet 4-6), i=" + i + "; val=" + val);
					pstmt.setNull(i + 1, java.sql.Types.NULL);
				}
				if (getDebug()) getLogger().debug("SlaveProcessor::insertRow: sql(4-5-4) END i=" + i);
			}
			if (getDebug()) getLogger().debug("SlaveProcessor::insertRow: sql(6), before pstmt.executeUpdate()");
			rc = pstmt.executeUpdate();
			time = System.currentTimeMillis() - time;
			if (time < 10) {
				getLogger().debug("SlaveProcessor::insertRow: INSERTED " + rc + " ROWS in " + time + " millisec.");
			} else {
				getLogger().warn("SlaveProcessor::insertRow: INSERTED " + rc + " ROWS in " + time + " millisec.");
			}
		} catch (Exception e) {
			throw new RservException("SlaveProcessor::insertRow: " + e.toString());
		} finally {
			//generalClose(pstmt, null);
		}

		return rc;
	}


	/*
	*/

	private int doInsert(SnapshotDataBlock snapDataBlock)
			throws RservException, RservDuplicateRowException {
		int rc = 0;
		try {
			if (snapDataBlock.getCommandInt() != CommandType.INSERT) {
				return rc;
			}

			TableMap tableMap = getTableMap();
			String tabname = snapDataBlock.getParam();
			//if (getDebug())
			getLogger().debug("SlaveProcessor::doInsert: table tabname=" + tabname);

			// NEW for ASYMMETRIC slave update:
			if (!tableMap.rowExists(tabname)) {
				getLogger().debug("SlaveProcessor::doInsert: table " +
						tabname + " does not exists - return 0");
				return 0;
			}
			int rc1 = 0;
			for (int i = 0; i < snapDataBlock.size(); i++) {
				try {
					rc1 = insertRow(tabname, snapDataBlock, i);
				} catch (Exception e) { // Sometimes row gets inserted before if rollbvack fails - try update
					String msg = e.getMessage();
					if (msg.indexOf("duplicate") >= 0) {
						rollback();
						rc1 = deleteDuplicateRow(tabname, snapDataBlock, i);
						commit();
					}
					throw new RservDuplicateRowException("Slave:: found duplicate");
				}
				if (rc1 > 0) {
					rc += rc1;
				}
			}
		} catch (RservDuplicateRowException re) {
			throw re;
		} catch (Exception e) {
			throw new RservException(
					"SlaveProcessor::doInsert: " + e.toString());
		}

		return rc;
	}


	/*
	*/

	protected int applySnapshot(SnapshotData snapData)
			throws RservException, RservDuplicateRowException {
		getLogger().debug("SlaveProcessor::applySnapshot: BEGIN");
		int rc = 0;

		setConstraintsDeferred();  // 7-16-2001 GK

		for (int i = 0; i < snapData.size(); i++) {
			SnapshotDataBlock snapDataBlock = snapData.getBlock(i);
			getLogger().debug("SlaveProcessor::applySnapshot: snapData.getBlock(i), i=" + i + "; snapDataBlock=" + snapDataBlock);

			int rc1 = doSyncId(snapDataBlock);
			if (getDebug()) getLogger().debug("SlaveProcessor::applySnapshot: after doSyncId, rc1=" + rc1);

			rc += doDelete(snapDataBlock);
			if (getDebug()) getLogger().debug("SlaveProcessor::applySnapshot: after doDelete, rc=" + rc);

			rc += doUpdate(snapDataBlock);
			if (getDebug()) getLogger().debug("SlaveProcessor::applySnapshot: after doUpdate, rc=" + rc);

			//rc += doInsert(snapDataBlock);
			//getLogger().debug("SlaveProcessor::applySnapshot: after doInsert, rc="+rc);
		}
		return rc;
	}


	public void resetTableColumnCache() {
		super.resetTableColumnCache();
		resetTableMap = true;
	}

}
