TransactionManager

package org.kuali.ole.docstore.transaction;

import org.apache.commons.lang.time.StopWatch;
import org.kuali.ole.RepositoryManager;
import org.kuali.ole.docstore.OleDocStoreException;
import org.kuali.ole.docstore.discovery.service.IndexerService;
import org.kuali.ole.docstore.document.DocumentManager;
import org.kuali.ole.docstore.document.WorkInstanceDocumentManager;
import org.kuali.ole.docstore.model.xmlpojo.ingest.RequestDocument;
import org.kuali.ole.docstore.model.xmlpojo.ingest.ResponseDocument;
import org.kuali.ole.docstore.model.xmlpojo.ingest.ResponseStatus;
import org.kuali.ole.docstore.process.ProcessParameters;
import org.kuali.ole.docstore.process.batch.BulkProcessRequest;
import org.kuali.ole.docstore.service.BeanLocator;
import org.kuali.ole.docstore.service.ServiceLocator;
import org.kuali.ole.docstore.utility.BatchIngestStatistics;
import org.kuali.ole.docstore.utility.BulkIngestStatistics;
import org.kuali.ole.pojo.OleException;
import org.kuali.ole.repository.NodeHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jcr.RepositoryException;
import javax.jcr.Session;
import java.io.FileNotFoundException;
import java.util.ArrayList;
import java.util.List;

/**
 * Encapsulates reusable common logic to support ACID transactions in DocStore.
 * User: tirumalesh.b
 * Date: 26/9/12 Time: 10:40 AM
 */
public class TransactionManager {
    private static final Logger logger = LoggerFactory.getLogger(NodeHandler.class);
    private IndexerService indexerService;
    private Session        session;
    private TransactionState      transactionState         = TransactionState.IDLE;
    private List<RequestDocument> newStateRequestDocuments = null;
    private List<RequestDocument> oldStateRequestDocuments = new ArrayList<RequestDocument>();

    public enum TransactionState {
        IDLE, STARTED, COMMITTED, ABORTED, FAILED
    }

    public void startSession(String user, String operation) throws OleDocStoreException {
        try {
            if (null == session) {
                session = RepositoryManager.getRepositoryManager().getSession(user, operation);
            }
        }
        catch (Exception e) {
            throw new OleDocStoreException(e);
        }
    }

    public void closeSession() {
        if (null != session) {
            try {
                RepositoryManager.getRepositoryManager().logout(session);
            }
            catch (OleException e) {
                logger.error(" Error while closing session :" + e);
            }
        }
    }

    public void startTransaction(String user, String operation) throws OleDocStoreException {
        if (transactionState == TransactionState.STARTED) {
            throw new OleDocStoreException("Transaction already started.");
        }
        try {
            if (null == session) {
                session = RepositoryManager.getRepositoryManager().getSession(user, operation);
            }
            if (null == indexerService) {
                indexerService = ServiceLocator.getIndexerService();
            }
        }
        catch (Exception e) {
            throw new OleDocStoreException(e);
        }
        transactionState = TransactionState.STARTED;
    }

    public void commit() throws OleDocStoreException {
        commit(null);
    }

    public void commit(BatchIngestStatistics batchIngestStatistics) throws OleDocStoreException {
        logger.info("Commit: Saving changes to index...");
        StopWatch solrCommitTimer = new StopWatch();
        try {
            if (null != batchIngestStatistics) {
                solrCommitTimer = new StopWatch();
                solrCommitTimer.start();
            }
            indexerService.commit();
            if (null != batchIngestStatistics) {
                solrCommitTimer.stop();
                batchIngestStatistics.setTimeToSolrCommit(solrCommitTimer.getTime());
            }
        }
        catch (Exception e) {
            transactionState = TransactionState.FAILED;
            logger.error("Exception during commit: Unable to save changes to index. :", e);
            try {
                session.refresh(false);
            }
            catch (Exception re) {
                // Ignore this, as data would not be saved anyway.
                logger.error("Exception during commit. Unable to rollback changes to docstore. :", re);
            }
            throw new OleDocStoreException("Commit failed.", e);
        }

        logger.info("Commit: Saving changes to docstore...");
        StopWatch sessionSaveTimer = null;
        try {
            if (null != batchIngestStatistics) {
                sessionSaveTimer = new StopWatch();
                sessionSaveTimer.start();
            }
            session.save();
            if (null != batchIngestStatistics) {
                sessionSaveTimer.stop();
                batchIngestStatistics.setTimeToSaveJcrSession(sessionSaveTimer.getTime());
            }
        }
        catch (Exception e) {
            transactionState = TransactionState.FAILED;
            logger.error("Exception during commit. Unable to save changes to docstore. :", e);
            logger.info("Commit: Reverting changes to index...");
            // TODO: Implement now. Delete data saved by indexerService.
            try {
                rollBackDataInIndexer(oldStateRequestDocuments);
//                rollBackDataInIndexer(oldStateRequestDocuments,newStateRequestDocuments);
            }
            catch (Exception ex) {
                logger.error("error while performing roll back in Indexer");
            }
            throw new OleDocStoreException("Commit failed.", e);
        }
        //Session logout is handled in DocumentServiceImpl.java
        // session.logout();
        // We are all done!!!
        transactionState = TransactionState.COMMITTED;
        newStateRequestDocuments = null;
        oldStateRequestDocuments = null;
    }

    private void rollBackDataInIndexer(List<RequestDocument> oldStateRequestDocuments) throws OleDocStoreException {
        if (oldStateRequestDocuments != null && oldStateRequestDocuments.size() > 0) {
            String result = indexerService.indexDocuments(oldStateRequestDocuments, false);
            if (!result.startsWith("success")) {
                throw new OleDocStoreException(result);
            }
        }
    }


   /* private void rollBackDataInIndexer(List<RequestDocument> newStateRequestDocuments,
                                       List<RequestDocument> oldStateRequestDocuments) throws OleDocStoreException {
        if (oldStateRequestDocuments != null && oldStateRequestDocuments.size() > 0) {
            String result = indexerService.indexDocuments(oldStateRequestDocuments, false);
            if (!result.startsWith("success")) {
                throw new OleDocStoreException(result);
            }
        }
        else if (newStateRequestDocuments != null && newStateRequestDocuments.size() > 0) {
            try {
                Map<String, List<String>> uuids = new HashMap<String, List<String>>();
                for (RequestDocument document : newStateRequestDocuments) {
                    for (RequestDocument linkedDoc : document.getLinkedRequestDocuments()) {
                        if (uuids.get(linkedDoc.getCategory()) == null) {
                            uuids.put(linkedDoc.getCategory(), new ArrayList<String>());
                        }
                        uuids.get(linkedDoc.getCategory()).add(linkedDoc.getUuid());
                    }
                    if (uuids.get(document.getCategory()) == null) {
                        uuids.put(document.getCategory(), new ArrayList<String>());
                    }
                    uuids.get(document.getCategory()).add(document.getUuid());
                }
                for (String category : uuids.keySet()) {
                    indexerService.deleteDocuments(category, uuids.get(category));
                }
            }
            catch (Exception e) {
                logger.info(e.getMessage(),e);
            }

        }
    }*/

    public void abort() {
        try {
            session.refresh(false);
        }
        catch (Exception re) {
            // Ignore this, as we want to rollback anyway.
            logger.error("Exception during abort. Unable to rollback changes to docstore. :", re);

        }
        try {
            indexerService.rollback();
        }
        catch (Exception e) {
            // Ignore this, as we want to rollback anyway.
            logger.error("Exception during abort. Unable to rollback changes to index. :", e);
        }
        transactionState = TransactionState.ABORTED;
        newStateRequestDocuments = null;
        oldStateRequestDocuments = null;
    }

    public void close() {
        session.logout();
        session = null;
        transactionState = TransactionState.IDLE;
        newStateRequestDocuments = null;
        oldStateRequestDocuments = null;
        setIndexerService(null);
    }

    public List<ResponseDocument> ingest(List<RequestDocument> requestDocuments) throws OleDocStoreException {
        DocumentManager documentManager = null;
        // Save the request documents so that they can be rolled back if necessary.
        newStateRequestDocuments = requestDocuments;
        List<ResponseDocument> responseDocuments = new ArrayList<ResponseDocument>();
        List<String> valuesList = new ArrayList<String>();
        for (RequestDocument requestDocument : requestDocuments) {
            // Each document could be of different cat-type-format.
            documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);
            ResponseDocument respDoc = new ResponseDocument();
            // validate given input

            try {
                documentManager.validateInput(requestDocument, session, valuesList);
            } catch (OleDocStoreException e) {
                respDoc.setStatus(ResponseStatus.INVALID_DATA.toString());
                logger.debug("validationMessage-->" + e.getMessage(),e);
                throw new OleDocStoreException(e.getMessage(),e);
            }

            // Store
            documentManager.ingest(requestDocument, session, respDoc);
            responseDocuments.add(respDoc);
        }
        // Index
        String result = indexerService.indexDocuments(requestDocuments, false);
        if (!result.startsWith("success")) {
            throw new OleDocStoreException(result);
        }

        return responseDocuments;
    }

    public ResponseDocument ingest(RequestDocument requestDocument) throws OleDocStoreException {
        DocumentManager documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);
        // Save the request documents so that they can be rolled back if necessary.
        newStateRequestDocuments = new ArrayList<RequestDocument>(1);
        newStateRequestDocuments.add(requestDocument);
        List<String> valuesList = new ArrayList<String>();
        // validate given input
        ResponseDocument respDoc = new ResponseDocument();
        try {
            documentManager.validateInput(requestDocument, session, valuesList);
        } catch (OleDocStoreException e) {
            respDoc.setStatus(ResponseStatus.INVALID_DATA.toString());
            logger.debug("validationMessage-->" + e.getMessage());
            throw new OleDocStoreException(e.getMessage());
        }
        documentManager.ingest(requestDocument, session, respDoc);
        String result = indexerService.indexDocument(requestDocument, false);
        if (!result.startsWith("success")) {
            throw new OleDocStoreException(result);
        }
        return respDoc;
    }


    public List<ResponseDocument> checkIn(List<RequestDocument> requestDocuments)
            throws OleDocStoreException {
        DocumentManager documentManager = null;
        // get the previous content from docStore to rollback the records if saving the session fails.

        List<ResponseDocument> responseDocuments = new ArrayList<ResponseDocument>();
        List<String> fieldValueList = new ArrayList<String>();
        for (RequestDocument requestDocument : requestDocuments) {
            // Each document could be of different cat-type-format.
            documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);
            if (requestDocument.getId() != null || requestDocument.getId().trim().length() > 0) {
                requestDocument.setUuid(requestDocument.getId());
            }
            //requestDocument.setOperation(operation);
            //requestDocument.setUser(user);
            // validate given input
            ResponseDocument respDoc = new ResponseDocument();

            try {
                documentManager.validateInput(requestDocument, session, fieldValueList);
            } catch (OleDocStoreException e) {
                respDoc.setStatus(ResponseStatus.INVALID_DATA.toString());
                logger.debug("validationMessage-->" + e.getMessage(),e);
                    throw new OleDocStoreException(e.getMessage());
            }
            // Each document could be of different cat-type-format.
            //documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);

            // update Docstore
            documentManager.checkin(requestDocument, session, respDoc);
            responseDocuments.add(respDoc);
        }
        // Index to solr
        String result = indexerService.indexDocuments(requestDocuments, false);
        if (!result.startsWith("success")) {
            throw new OleDocStoreException(result);
        }

        return responseDocuments;
    }


    public List<ResponseDocument> checkOut(List<RequestDocument> requestDocuments, String user)
            throws OleDocStoreException {
        DocumentManager documentManager = null;
        // Save the request documents so that they can be rolled back if necessary.
        newStateRequestDocuments = requestDocuments;
        List<ResponseDocument> responseDocuments = new ArrayList<ResponseDocument>();
        for (RequestDocument requestDocument : requestDocuments) {
            // Each document could be of different cat-type-format.
            documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);
            // To differentiate between restfull API and Normal CheckOut.
            requestDocument.setUser(user);
            // checkOut
            responseDocuments.add(documentManager.checkout(requestDocument, session));
        }
        return responseDocuments;
    }


    public ResponseDocument checkIn(RequestDocument requestDocument) throws OleDocStoreException {
        DocumentManager documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);
        // Save the request documents so that they can be rolled back if necessary.
        newStateRequestDocuments = new ArrayList<RequestDocument>(1);
        newStateRequestDocuments.add(requestDocument);
        ResponseDocument responseDocument = new ResponseDocument();
        List<String> fieldValueList = new ArrayList<String>();

        // validate given input
        ResponseDocument respDoc = new ResponseDocument();

        try {
            documentManager.validateInput(requestDocument, session, fieldValueList);
        } catch (OleDocStoreException e) {
            respDoc.setStatus(ResponseStatus.INVALID_DATA.toString());
            logger.debug("validationMessage-->" + e.getMessage(),e);
            throw new OleDocStoreException(e.getMessage());
        }
        // update Docstore
        documentManager.checkin(requestDocument, session, responseDocument);
        //Index to solr
        String result = indexerService.indexDocument(requestDocument, false);
        if (!result.startsWith("success")) {
            throw new OleDocStoreException(result);
        }
        return responseDocument;
    }


    public List<ResponseDocument> bind(List<RequestDocument> requestDocuments, String operation)
            throws Exception, RepositoryException, OleException, FileNotFoundException {
        DocumentManager documentManager = null;
        // get the previous content from docStore to rollback the records if saving the session fails.

        List<ResponseDocument> responseDocuments = new ArrayList<ResponseDocument>();
        for (RequestDocument requestDocument : requestDocuments) {
            documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);
            // update Docstore
            responseDocuments.add(documentManager.bind(requestDocument, session, operation));
        }

        // Index to solr
        String result = indexerService.bind(requestDocuments);
        if (!result.startsWith("success")) {
            throw new OleDocStoreException(result);
        }

        return responseDocuments;
    }


    public List<ResponseDocument> unbind(List<RequestDocument> requestDocuments, String operation)
            throws Exception, RepositoryException, OleException, FileNotFoundException {
        DocumentManager documentManager = null;
        // get the previous content from docStore to rollback the records if saving the session fails.

        List<ResponseDocument> responseDocuments = new ArrayList<ResponseDocument>();
        for (RequestDocument requestDocument : requestDocuments) {
            documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);
            // update Docstore
            responseDocuments.add(documentManager.unbind(requestDocument, session, operation));

        }

        // Index to solr
        String result = indexerService.unbind(requestDocuments);
        if (!result.startsWith("success")) {
            throw new OleDocStoreException(result);
        }

        return responseDocuments;
    }


    public void batchIngest(BulkProcessRequest bulkProcessRequest, List<RequestDocument> requestDocuments)
            throws OleDocStoreException {
        BulkIngestStatistics bulkIngestStatistics = bulkProcessRequest.getBulkIngestStatistics();
        BatchIngestStatistics batchStatistics = bulkIngestStatistics.getCurrentBatch();
        long commitSize = ProcessParameters.BULK_INGEST_COMMIT_SIZE;
        newStateRequestDocuments = new ArrayList<RequestDocument>();
        // Save the request documents so that they can be rolled back if necessary.
        newStateRequestDocuments.addAll(requestDocuments);
        DocumentManager documentManager = BeanLocator.getDocumentManagerFactory()
                                                     .getDocumentManager(requestDocuments.get(0));

        // Store
        StopWatch createNodesTimer = new StopWatch();
        createNodesTimer.start();
        documentManager.ingest(requestDocuments, session);
        createNodesTimer.stop();
        batchStatistics.setTimeToCreateNodesInJcr(createNodesTimer.getTime());
        batchStatistics.setCommitSize(commitSize);

        StopWatch indexSolrDocsTime = new StopWatch();
        indexSolrDocsTime.start();

        // Index
        String result = indexerService.indexDocuments(requestDocuments, false);
        indexSolrDocsTime.stop();
        batchStatistics.setTimeToIndexSolrInputDocs(indexSolrDocsTime.getTime());

        if (!result.startsWith("success")) {
            throw new OleDocStoreException(result);
        }
    }

    public List<ResponseDocument> delete(List<RequestDocument> requestDocuments) throws Exception{
        DocumentManager documentManager = null;
        // Save the request documents so that they can be rolled back if necessary.
        newStateRequestDocuments = requestDocuments;
        List<ResponseDocument> responseDocuments = new ArrayList<ResponseDocument>();
        for (RequestDocument requestDocument : requestDocuments) {
            // Each document could be of different cat-type-format.
            documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);
            //Store
            responseDocuments.add(documentManager.delete(requestDocument,session));
        }
        // Index
        String result = indexerService.delete(requestDocuments);
        if (!result.startsWith("success")) {
            throw new OleDocStoreException(result);
        }
        return responseDocuments;
    }

    public List<ResponseDocument> deleteVerify(List<RequestDocument> requestDocuments) throws Exception {
        DocumentManager documentManager = null;
        List<ResponseDocument> responseDocuments = new ArrayList<ResponseDocument>();
        for (RequestDocument requestDocument : requestDocuments) {
            documentManager = BeanLocator.getDocumentManagerFactory().getDocumentManager(requestDocument);
            responseDocuments.add(documentManager.deleteVerify(requestDocument, session));
        }
        return responseDocuments;
    }
    public IndexerService getIndexerService() {
        return indexerService;
    }

    public void setIndexerService(IndexerService indexerService) {
        this.indexerService = indexerService;
    }
       public void transferInstances(List<RequestDocument> requestDocuments) throws Exception {
           WorkInstanceDocumentManager.getInstance().transferInstances(requestDocuments,session);
        indexerService.transferInstances(requestDocuments);

    }
    public void transferItems(List<RequestDocument> requestDocuments) throws Exception {
        logger.debug("TransactionManager transferItems");
        WorkInstanceDocumentManager.getInstance().transferItems(requestDocuments, session);
        indexerService.transferItems(requestDocuments);

    }
}