Metacat

/**

 *  '$RCSfile$'

 *  Copyright: 2000-2011 Regents of the University of California and the

 *              National Center for Ecological Analysis and Synthesis

 *

 *   '$Author:  $'

 *     '$Date:  $'

 *

 * 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

 */

package edu.ucsb.nceas.metacat.dataone;

import java.io.ByteArrayInputStream;

import java.io.ByteArrayOutputStream;

import java.io.File;

import java.io.FileInputStream;

import java.io.FileOutputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.InputStreamReader;

import java.io.OutputStreamWriter;

import java.io.UnsupportedEncodingException;

import java.io.Writer;

import java.math.BigInteger;

import java.net.URISyntaxException;

import java.security.NoSuchAlgorithmException;

import java.sql.SQLException;

import java.util.ArrayList;

import java.util.Calendar;

import java.util.Date;

import java.util.HashMap;

import java.util.HashSet;

import java.util.Hashtable;

import java.util.List;

import java.util.Map;

import java.util.Set;

import java.util.Timer;

import java.util.UUID;

import java.util.Vector;

import javax.servlet.http.HttpServletRequest;

import org.apache.commons.io.IOUtils;

import org.apache.log4j.Logger;

import org.dataone.client.v2.CNode;

import org.dataone.client.v2.itk.D1Client;

import org.dataone.client.v2.MNode;

import org.dataone.client.v2.formats.ObjectFormatCache;

import org.dataone.client.auth.CertificateManager;

import org.dataone.client.v2.formats.ObjectFormatInfo;

import org.dataone.configuration.Settings;

import org.dataone.ore.ResourceMapFactory;

import org.dataone.service.exceptions.BaseException;

import org.dataone.service.exceptions.IdentifierNotUnique;

import org.dataone.service.exceptions.InsufficientResources;

import org.dataone.service.exceptions.InvalidRequest;

import org.dataone.service.exceptions.InvalidSystemMetadata;

import org.dataone.service.exceptions.InvalidToken;

import org.dataone.service.exceptions.NotAuthorized;

import org.dataone.service.exceptions.NotFound;

import org.dataone.service.exceptions.NotImplemented;

import org.dataone.service.exceptions.ServiceFailure;

import org.dataone.service.exceptions.SynchronizationFailed;

import org.dataone.service.exceptions.UnsupportedType;

import org.dataone.service.mn.tier1.v2.MNCore;

import org.dataone.service.mn.tier1.v2.MNRead;

import org.dataone.service.mn.tier2.v2.MNAuthorization;

import org.dataone.service.mn.tier3.v2.MNStorage;

import org.dataone.service.mn.tier4.v2.MNReplication;

import org.dataone.service.mn.v2.MNPackage;

import org.dataone.service.mn.v2.MNQuery;

import org.dataone.service.mn.v2.MNView;

import org.dataone.service.types.v1.AccessRule;

import org.dataone.service.types.v1.Checksum;

import org.dataone.service.types.v1.DescribeResponse;

import org.dataone.service.types.v1.Event;

import org.dataone.service.types.v1.Identifier;

import org.dataone.service.types.v2.Log;

import org.dataone.service.types.v2.LogEntry;

import org.dataone.service.types.v2.OptionList;

import org.dataone.service.types.v1.MonitorInfo;

import org.dataone.service.types.v1.MonitorList;

import org.dataone.service.types.v2.Node;

import org.dataone.service.types.v2.NodeList;

import org.dataone.service.types.v1.NodeReference;

import org.dataone.service.types.v1.NodeState;

import org.dataone.service.types.v1.NodeType;

import org.dataone.service.types.v2.ObjectFormat;

import org.dataone.service.types.v1.ObjectFormatIdentifier;

import org.dataone.service.types.v1.ObjectList;

import org.dataone.service.types.v1.Permission;

import org.dataone.service.types.v1.Ping;

import org.dataone.service.types.v1.ReplicationStatus;

import org.dataone.service.types.v1.Schedule;

import org.dataone.service.types.v1.Service;

import org.dataone.service.types.v1.Services;

import org.dataone.service.types.v1.Session;

import org.dataone.service.types.v1.Subject;

import org.dataone.service.types.v1.Synchronization;

import org.dataone.service.types.v2.SystemMetadata;

import org.dataone.service.types.v1.util.AuthUtils;

import org.dataone.service.types.v1.util.ChecksumUtil;

import org.dataone.service.types.v1_1.QueryEngineDescription;

import org.dataone.service.types.v1_1.QueryEngineList;

import org.dataone.service.types.v1_1.QueryField;

import org.dataone.service.util.Constants;

import org.dataone.service.util.TypeMarshaller;

import org.dspace.foresite.OREException;

import org.dspace.foresite.OREParserException;

import org.dspace.foresite.ORESerialiserException;

import org.dspace.foresite.ResourceMap;

import org.ecoinformatics.datamanager.parser.DataPackage;

import org.ecoinformatics.datamanager.parser.Entity;

import org.ecoinformatics.datamanager.parser.generic.DataPackageParserInterface;

import org.ecoinformatics.datamanager.parser.generic.Eml200DataPackageParser;

import edu.ucsb.nceas.ezid.EZIDException;

import edu.ucsb.nceas.metacat.DBQuery;

import edu.ucsb.nceas.metacat.DBTransform;

import edu.ucsb.nceas.metacat.EventLog;

import edu.ucsb.nceas.metacat.IdentifierManager;

import edu.ucsb.nceas.metacat.McdbDocNotFoundException;

import edu.ucsb.nceas.metacat.MetaCatServlet;

import edu.ucsb.nceas.metacat.MetacatHandler;

import edu.ucsb.nceas.metacat.common.query.EnabledQueryEngines;

import edu.ucsb.nceas.metacat.common.query.stream.ContentTypeByteArrayInputStream;

import edu.ucsb.nceas.metacat.dataone.hazelcast.HazelcastService;

import edu.ucsb.nceas.metacat.index.MetacatSolrEngineDescriptionHandler;

import edu.ucsb.nceas.metacat.index.MetacatSolrIndex;

import edu.ucsb.nceas.metacat.properties.PropertyService;

import edu.ucsb.nceas.metacat.shared.MetacatUtilException;

import edu.ucsb.nceas.metacat.util.DeleteOnCloseFileInputStream;

import edu.ucsb.nceas.metacat.util.DocumentUtil;

import edu.ucsb.nceas.metacat.util.SkinUtil;

import edu.ucsb.nceas.metacat.util.SystemUtil;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

import edu.ucsb.nceas.utilities.XMLUtilities;

import edu.ucsb.nceas.utilities.export.HtmlToPdf;

import gov.loc.repository.bagit.Bag;

import gov.loc.repository.bagit.BagFactory;

import gov.loc.repository.bagit.writer.impl.ZipWriter;

/**

 * Represents Metacat's implementation of the DataONE Member Node 

 * service API. Methods implement the various MN* interfaces, and methods common

 * to both Member Node and Coordinating Node interfaces are found in the

 * D1NodeService base class.

 * 

 * Implements:

 * MNCore.ping()

 * MNCore.getLogRecords()

 * MNCore.getObjectStatistics()

 * MNCore.getOperationStatistics()

 * MNCore.getStatus()

 * MNCore.getCapabilities()

 * MNRead.get()

 * MNRead.getSystemMetadata()

 * MNRead.describe()

 * MNRead.getChecksum()

 * MNRead.listObjects()

 * MNRead.synchronizationFailed()

 * MNAuthorization.isAuthorized()

 * MNAuthorization.setAccessPolicy()

 * MNStorage.create()

 * MNStorage.update()

 * MNStorage.delete()

 * MNStorage.updateSystemMetadata()

 * MNReplication.replicate()

 * 

 */

public class MNodeService extends D1NodeService 

    implements MNAuthorization, MNCore, MNRead, MNReplication, MNStorage, MNQuery, MNView, MNPackage {

    //private static final String PATHQUERY = "pathquery";

	public static final String UUID_SCHEME = "UUID";

	public static final String DOI_SCHEME = "DOI";

	private static final String UUID_PREFIX = "urn:uuid:";

	/* the logger instance */

    private Logger logMetacat = null;

    /* A reference to a remote Memeber Node */

    private MNode mn;

    /* A reference to a Coordinating Node */

    private CNode cn;

    /**

     * Singleton accessor to get an instance of MNodeService.

     * 

     * @return instance - the instance of MNodeService

     */

    public static MNodeService getInstance(HttpServletRequest request) {

        return new MNodeService(request);

    }

    /**

     * Constructor, private for singleton access

     */

    private MNodeService(HttpServletRequest request) {

        super(request);

        logMetacat = Logger.getLogger(MNodeService.class);

        // set the Member Node certificate file location

        CertificateManager.getInstance().setCertificateLocation(Settings.getConfiguration().getString("D1Client.certificate.file"));

    }

    /**

     * Deletes an object from the Member Node, where the object is either a 

     * data object or a science metadata object.

     * 

     * @param session - the Session object containing the credentials for the Subject

     * @param pid - The object identifier to be deleted

     * 

     * @return pid - the identifier of the object used for the deletion

     * 

     * @throws InvalidToken

     * @throws ServiceFailure

     * @throws NotAuthorized

     * @throws NotFound

     * @throws NotImplemented

     * @throws InvalidRequest

     */

    @Override

    public Identifier delete(Session session, Identifier pid) 

        throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented {

    	// only admin of  the MN or the CN is allowed a full delete

        boolean allowed = false;

        allowed = isAdminAuthorized(session);

        String serviceFailureCode = "2902";

        Identifier sid = getPIDForSID(pid, serviceFailureCode);

        if(sid != null) {

            pid = sid;

        }

        //check if it is the authoritative member node

        if(!allowed) {

            allowed = isAuthoritativeMNodeAdmin(session, pid);

        }

        if (!allowed) { 

            throw new NotAuthorized("1320", "The provided identity does not have " + "permission to delete objects on the Node.");

        }

    	// defer to superclass implementation

        return super.delete(session, pid);

    }

    /**

     * Updates an existing object by creating a new object identified by 

     * newPid on the Member Node which explicitly obsoletes the object 

     * identified by pid through appropriate changes to the SystemMetadata 

     * of pid and newPid

     * 

     * @param session - the Session object containing the credentials for the Subject

     * @param pid - The identifier of the object to be updated

     * @param object - the new object bytes

     * @param sysmeta - the new system metadata describing the object

     * 

     * @return newPid - the identifier of the new object

     * 

     * @throws InvalidToken

     * @throws ServiceFailure

     * @throws NotAuthorized

     * @throws NotFound

     * @throws NotImplemented

     * @throws IdentifierNotUnique

     * @throws UnsupportedType

     * @throws InsufficientResources

     * @throws InvalidSystemMetadata

     * @throws InvalidRequest

     */

    @Override

    public Identifier update(Session session, Identifier pid, InputStream object, 

        Identifier newPid, SystemMetadata sysmeta) 

        throws InvalidToken, ServiceFailure, NotAuthorized, IdentifierNotUnique, 

        UnsupportedType, InsufficientResources, NotFound, 

        InvalidSystemMetadata, NotImplemented, InvalidRequest {

        //transform a sid to a pid if it is applicable

        String serviceFailureCode = "1310";

        Identifier sid = getPIDForSID(pid, serviceFailureCode);

        if(sid != null) {

            pid = sid;

        }

        String localId = null;

        boolean allowed = false;

        boolean isScienceMetadata = false;

        if (session == null) {

        	throw new InvalidToken("1210", "No session has been provided");

        }

        Subject subject = session.getSubject();

        // verify the pid is valid format

        if (!isValidIdentifier(pid)) {

        	throw new InvalidRequest("1202", "The provided identifier is invalid.");

        }

        // verify the new pid is valid format

        if (!isValidIdentifier(newPid)) {

            throw new InvalidRequest("1202", "The provided identifier is invalid.");

        }

        // make sure that the newPid doesn't exists

        boolean idExists = true;

        try {

            idExists = IdentifierManager.getInstance().identifierExists(newPid.getValue());

        } catch (SQLException e) {

            throw new ServiceFailure("1310", 

                                    "The requested identifier " + newPid.getValue() +

                                    " couldn't be determined if it is unique since : "+e.getMessage());

        }

        if (idExists) {

                throw new IdentifierNotUnique("1220", 

                          "The requested identifier " + newPid.getValue() +

                          " is already used by another object and" +

                          "therefore can not be used for this object. Clients should choose" +

                          "a new identifier that is unique and retry the operation or " +

                          "use CN.reserveIdentifier() to reserve one.");

        }

        // check for the existing identifier

        try {

            localId = IdentifierManager.getInstance().getLocalId(pid.getValue());

        } catch (McdbDocNotFoundException e) {

            throw new InvalidRequest("1202", "The object with the provided " + 

                "identifier was not found.");

        } catch (SQLException ee) {

            throw new ServiceFailure("1310", "The object with the provided " + 

                    "identifier "+pid.getValue()+" can't be identified since - "+ee.getMessage());

        }

        // set the originating node

        NodeReference originMemberNode = this.getCapabilities().getIdentifier();

        sysmeta.setOriginMemberNode(originMemberNode);

        // set the submitter to match the certificate

        sysmeta.setSubmitter(subject);

        // set the dates

        Date now = Calendar.getInstance().getTime();

        sysmeta.setDateSysMetadataModified(now);

        sysmeta.setDateUploaded(now);

        // make sure serial version is set to something

        BigInteger serialVersion = sysmeta.getSerialVersion();

        if (serialVersion == null) {

        	sysmeta.setSerialVersion(BigInteger.ZERO);

        }

        // does the subject have WRITE ( == update) priveleges on the pid?

        //allowed = isAuthorized(session, pid, Permission.WRITE);

        //CN having the permission is allowed; user with the write permission and calling on the authoritative node is allowed.

        allowed = allowUpdating(session, pid, Permission.WRITE);

        if (allowed) {

        	// check quality of SM

        	if (sysmeta.getObsoletedBy() != null) {

        		throw new InvalidSystemMetadata("1300", "Cannot include obsoletedBy when updating object");

        	}

        	if (sysmeta.getObsoletes() != null && !sysmeta.getObsoletes().getValue().equals(pid.getValue())) {

        		throw new InvalidSystemMetadata("1300", "The identifier provided in obsoletes does not match old Identifier");

        	}

            // get the existing system metadata for the object

            SystemMetadata existingSysMeta = getSystemMetadata(session, pid);

            //System.out.println("the archive is "+existingSysMeta.getArchived());

            //Base on documentation, we can't update an archived object:

            //The update operation MUST fail with Exceptions.InvalidRequest on objects that have the Types.SystemMetadata.archived property set to true.

            if(existingSysMeta.getArchived() != null && existingSysMeta.getArchived()) {

                throw new InvalidRequest("1202","An archived object"+pid.getValue()+" can't be updated");

            }

            // check for previous update

            // see: https://redmine.dataone.org/issues/3336

            Identifier existingObsoletedBy = existingSysMeta.getObsoletedBy();

            if (existingObsoletedBy != null) {

            	throw new InvalidRequest("1202", 

            			"The previous identifier has already been made obsolete by: " + existingObsoletedBy.getValue());

            }

            //check the sid in the system metadata. If it exists, it should be non-exist or match the old sid in the previous system metadata.

            Identifier sidInSys = sysmeta.getSeriesId();

            if(sidInSys != null) {

                if (!isValidIdentifier(sidInSys)) {

                    throw new InvalidSystemMetadata("1300", "The provided series id in the system metadata is invalid.");

                }

                Identifier previousSid = existingSysMeta.getSeriesId();

                if(previousSid != null) {

                    // there is a previous sid, if the new sid doesn't match it, the new sid should be non-existing.

                    if(!sidInSys.getValue().equals(previousSid.getValue())) {

                        try {

                            idExists = IdentifierManager.getInstance().identifierExists(sidInSys.getValue());

                        } catch (SQLException e) {

                            throw new ServiceFailure("1310", 

                                                    "The requested identifier " + sidInSys.getValue() +

                                                    " couldn't be determined if it is unique since : "+e.getMessage());

                        }

                        if(idExists) {

                            throw new InvalidSystemMetadata("1300", "The series id "+sidInSys.getValue()+" in the system metadata doesn't match the previous series id "

                                                            +previousSid.getValue()+", so it should NOT exist. However, it was used by another object.");

                        }

                    }

                } else {

                    // there is no previous sid, the new sid should be non-existing.

                    try {

                        idExists = IdentifierManager.getInstance().identifierExists(sidInSys.getValue());

                    } catch (SQLException e) {

                        throw new ServiceFailure("1310", 

                                                "The requested identifier " + sidInSys.getValue() +

                                                " couldn't be determined if it is unique since : "+e.getMessage());

                    }

                    if(idExists) {

                        throw new InvalidSystemMetadata("1300", "The series id "+sidInSys.getValue()+" in the system metadata should NOT exist since the previous series id is null."

                                                        +"However, it was used by another object.");

                    }

                }

                //the series id equals the pid (new pid hasn't been registered in the system, so IdentifierManager.getInstance().identifierExists method can't exclude this scenario)

                if(sidInSys.getValue().equals(newPid.getValue())) {

                    throw new InvalidSystemMetadata("1300", "The series id "+sidInSys.getValue()+" in the system metadata shouldn't have the same value of the pid.");

                }

            }

            isScienceMetadata = isScienceMetadata(sysmeta);

            // do we have XML metadata or a data object?

            if (isScienceMetadata) {

                // update the science metadata XML document

                // TODO: handle non-XML metadata/data documents (like netCDF)

                // TODO: don't put objects into memory using stream to string

                //String objectAsXML = "";

                try {

                    //objectAsXML = IOUtils.toString(object, "UTF-8");

                    // give the old pid so we can calculate the new local id 

                    localId = insertOrUpdateDocument(object, "UTF-8", pid, session, "update");

                    // register the newPid and the generated localId

                    if (newPid != null) {

                        IdentifierManager.getInstance().createMapping(newPid.getValue(), localId);

                    }

                } catch (IOException e) {

                    String msg = "The Node is unable to create the object. " + "There was a problem converting the object to XML";

                    logMetacat.info(msg);

                    throw new ServiceFailure("1310", msg + ": " + e.getMessage());

                }

            } else {

                // update the data object

                localId = insertDataObject(object, newPid, session);

            }

            // add the newPid to the obsoletedBy list for the existing sysmeta

            existingSysMeta.setObsoletedBy(newPid);

            // then update the existing system metadata

            updateSystemMetadata(existingSysMeta);

            // prep the new system metadata, add pid to the affected lists

            sysmeta.setObsoletes(pid);

            //sysmeta.addDerivedFrom(pid);

            // and insert the new system metadata

            insertSystemMetadata(sysmeta);

            // log the update event

            EventLog.getInstance().log(request.getRemoteAddr(), request.getHeader("User-Agent"), subject.getValue(), localId, Event.UPDATE.toString());

            // attempt to register the identifier - it checks if it is a doi

            try {

    			DOIService.getInstance().registerDOI(sysmeta);

    		} catch (Exception e) {

                throw new ServiceFailure("1190", "Could not register DOI: " + e.getMessage());

    		}

        } else {

            throw new NotAuthorized("1200", "The provided identity does not have " + "permission to UPDATE the object identified by " + pid.getValue()

                    + " on the Member Node.");

        }

        return newPid;

    }

    public Identifier create(Session session, Identifier pid, InputStream object, SystemMetadata sysmeta) throws InvalidToken, ServiceFailure, NotAuthorized,

            IdentifierNotUnique, UnsupportedType, InsufficientResources, InvalidSystemMetadata, NotImplemented, InvalidRequest {

        // check for null session

        if (session == null) {

          throw new InvalidToken("1110", "Session is required to WRITE to the Node.");

        }

        // verify the pid is valid format

        if (!isValidIdentifier(pid)) {

            throw new InvalidRequest("1102", "The provided identifier is invalid.");

        }

        // set the submitter to match the certificate

        sysmeta.setSubmitter(session.getSubject());

        // set the originating node

        NodeReference originMemberNode = this.getCapabilities().getIdentifier();

        sysmeta.setOriginMemberNode(originMemberNode);

        sysmeta.setArchived(false);

        // set the dates

        Date now = Calendar.getInstance().getTime();

        sysmeta.setDateSysMetadataModified(now);

        sysmeta.setDateUploaded(now);

        // set the serial version

        sysmeta.setSerialVersion(BigInteger.ZERO);

        // check that we are not attempting to subvert versioning

        if (sysmeta.getObsoletes() != null && sysmeta.getObsoletes().getValue() != null) {

            throw new InvalidSystemMetadata("1180", 

              "The supplied system metadata is invalid. " +

              "The obsoletes field cannot have a value when creating entries.");

        }

        if (sysmeta.getObsoletedBy() != null && sysmeta.getObsoletedBy().getValue() != null) {

            throw new InvalidSystemMetadata("1180", 

              "The supplied system metadata is invalid. " +

              "The obsoletedBy field cannot have a value when creating entries.");

        }

        // verify the sid in the system metadata

        Identifier sid = sysmeta.getSeriesId();

        boolean idExists = false;

        if(sid != null) {

            if (!isValidIdentifier(sid)) {

                throw new InvalidSystemMetadata("1180", "The provided series id is invalid.");

            }

            try {

                idExists = IdentifierManager.getInstance().identifierExists(sid.getValue());

            } catch (SQLException e) {

                throw new ServiceFailure("1190", 

                                        "The series identifier " + sid.getValue() +

                                        " in the system metadata couldn't be determined if it is unique since : "+e.getMessage());

            }

            if (idExists) {

                    throw new InvalidSystemMetadata("1180", 

                              "The series identifier " + sid.getValue() +

                              " is already used by another object and" +

                              "therefore can not be used for this object. Clients should choose" +

                              "a new identifier that is unique and retry the operation or " +

                              "use CN.reserveIdentifier() to reserve one.");

            }

            //the series id equals the pid (new pid hasn't been registered in the system, so IdentifierManager.getInstance().identifierExists method can't exclude this scenario )

            if(sid.getValue().equals(pid.getValue())) {

                throw new InvalidSystemMetadata("1180", "The series id "+sid.getValue()+" in the system metadata shouldn't have the same value of the pid.");

            }

        }

        // call the shared impl

        Identifier resultPid = super.create(session, pid, object, sysmeta);

        // attempt to register the identifier - it checks if it is a doi

        try {

			DOIService.getInstance().registerDOI(sysmeta);

		} catch (Exception e) {

			ServiceFailure sf = new ServiceFailure("1190", "Could not register DOI: " + e.getMessage());

			sf.initCause(e);

            throw sf;

		}

        // return 

		return resultPid ;

    }

    /**

     * Called by a Coordinating Node to request that the Member Node create a 

     * copy of the specified object by retrieving it from another Member 

     * Node and storing it locally so that it can be made accessible to 

     * the DataONE system.

     * 

     * @param session - the Session object containing the credentials for the Subject

     * @param sysmeta - Copy of the CN held system metadata for the object

     * @param sourceNode - A reference to node from which the content should be 

     *                     retrieved. The reference should be resolved by 

     *                     checking the CN node registry.

     * 

     * @return true if the replication succeeds

     * 

     * @throws ServiceFailure

     * @throws NotAuthorized

     * @throws NotImplemented

     * @throws UnsupportedType

     * @throws InsufficientResources

     * @throws InvalidRequest

     */

    @Override

    public boolean replicate(Session session, SystemMetadata sysmeta,

            NodeReference sourceNode) throws NotImplemented, ServiceFailure,

            NotAuthorized, InvalidRequest, InsufficientResources,

            UnsupportedType {

        if (session != null && sysmeta != null && sourceNode != null) {

            logMetacat.info("MNodeService.replicate() called with parameters: \n" +

                            "\tSession.Subject      = "                           +

                            session.getSubject().getValue() + "\n"                +

                            "\tidentifier           = "                           + 

                            sysmeta.getIdentifier().getValue()                    +

                            "\n" + "\tSource NodeReference ="                     +

                            sourceNode.getValue());

        }

        boolean result = false;

        String nodeIdStr = null;

        NodeReference nodeId = null;

        // get the referenced object

        Identifier pid = sysmeta.getIdentifier();

        // verify the pid is valid format

        if (!isValidIdentifier(pid)) {

            throw new InvalidRequest("2153", "The provided identifier in the system metadata is invalid.");

        }

        // get from the membernode

        // TODO: switch credentials for the server retrieval?

        this.mn = D1Client.getMN(sourceNode);

        this.cn = D1Client.getCN();

        InputStream object = null;

        Session thisNodeSession = null;

        SystemMetadata localSystemMetadata = null;

        BaseException failure = null;

        String localId = null;

        // TODO: check credentials

        // cannot be called by public

        if (session == null || session.getSubject() == null) {

            String msg = "No session was provided to replicate identifier " +

            sysmeta.getIdentifier().getValue();

            logMetacat.error(msg);

            throw new NotAuthorized("2152", msg);

        }

        // get the local node id

        try {

            nodeIdStr = PropertyService.getProperty("dataone.nodeId");

            nodeId = new NodeReference();

            nodeId.setValue(nodeIdStr);

        } catch (PropertyNotFoundException e1) {

            String msg = "Couldn't get dataone.nodeId property: " + e1.getMessage();

            failure = new ServiceFailure("2151", msg);

            //setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, failure);

            logMetacat.error(msg);

            //return true;

            throw new ServiceFailure("2151", msg);

        }

        try {

            // do we already have a replica?

            try {

                localId = IdentifierManager.getInstance().getLocalId(pid.getValue());

                // if we have a local id, get the local object

                try {

                    object = MetacatHandler.read(localId);

                } catch (Exception e) {

                	// NOTE: we may already know about this ID because it could be a data file described by a metadata file

                	// https://redmine.dataone.org/issues/2572

                	// TODO: fix this so that we don't prevent ourselves from getting replicas

                    // let the CN know that the replication failed

                	logMetacat.warn("Object content not found on this node despite having localId: " + localId);

                	String msg = "Can't read the object bytes properly, replica is invalid.";

                    ServiceFailure serviceFailure = new ServiceFailure("2151", msg);

                    setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, serviceFailure);

                    logMetacat.warn(msg);

                    throw serviceFailure;

                }

            } catch (McdbDocNotFoundException e) {

                logMetacat.info("No replica found. Continuing.");

            } catch (SQLException ee) {

                throw new ServiceFailure("2151", "Couldn't identify the local id of the object with the specified identifier "

                                        +pid.getValue()+" since - "+ee.getMessage());

            }

            // no local replica, get a replica

            if ( object == null ) {

                // session should be null to use the default certificate

                // location set in the Certificate manager

                object = mn.getReplica(thisNodeSession, pid);

                logMetacat.info("MNodeService.getReplica() called for identifier "

                                + pid.getValue());

            }

        } catch (InvalidToken e) {            

            String msg = "Could not retrieve object to replicate (InvalidToken): "+ e.getMessage();

            failure = new ServiceFailure("2151", msg);

            setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, failure);

            logMetacat.error(msg);

            throw new ServiceFailure("2151", msg);

        } catch (NotFound e) {

            String msg = "Could not retrieve object to replicate (NotFound): "+ e.getMessage();

            failure = new ServiceFailure("2151", msg);

            setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, failure);

            logMetacat.error(msg);

            throw new ServiceFailure("2151", msg);

        } catch (NotAuthorized e) {

            String msg = "Could not retrieve object to replicate (NotAuthorized): "+ e.getMessage();

            failure = new ServiceFailure("2151", msg);

            setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, failure);

            logMetacat.error(msg);

            throw new ServiceFailure("2151", msg);

        } catch (NotImplemented e) {

            String msg = "Could not retrieve object to replicate (mn.getReplica NotImplemented): "+ e.getMessage();

            failure = new ServiceFailure("2151", msg);

            setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, failure);

            logMetacat.error(msg);

            throw new ServiceFailure("2151", msg);

        } catch (ServiceFailure e) {

            String msg = "Could not retrieve object to replicate (ServiceFailure): "+ e.getMessage();

            failure = new ServiceFailure("2151", msg);

            setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, failure);

            logMetacat.error(msg);

            throw new ServiceFailure("2151", msg);

        } catch (InsufficientResources e) {

            String msg = "Could not retrieve object to replicate (InsufficientResources): "+ e.getMessage();

            failure = new ServiceFailure("2151", msg);

            setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, failure);

            logMetacat.error(msg);

            throw new ServiceFailure("2151", msg);

        }

        // verify checksum on the object, if supported

        if (object.markSupported()) {

            Checksum givenChecksum = sysmeta.getChecksum();

            Checksum computedChecksum = null;

            try {

                computedChecksum = ChecksumUtil.checksum(object, givenChecksum.getAlgorithm());

                object.reset();

            } catch (Exception e) {

                String msg = "Error computing checksum on replica: " + e.getMessage();

                logMetacat.error(msg);

                ServiceFailure sf = new ServiceFailure("2151", msg);

                sf.initCause(e);

                setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, sf);

                throw sf;

            }

            if (!givenChecksum.getValue().equals(computedChecksum.getValue())) {

                logMetacat.error("Given    checksum for " + pid.getValue() + 

                    "is " + givenChecksum.getValue());

                logMetacat.error("Computed checksum for " + pid.getValue() + 

                    "is " + computedChecksum.getValue());

                String msg = "Computed checksum does not match declared checksum";

                failure = new ServiceFailure("2151", msg);

                setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, failure);

                throw new ServiceFailure("2151", msg);

            }

        }

        // add it to local store

        Identifier retPid;

        try {

            // skip the MN.create -- this mutates the system metadata and we don't want it to

            if ( localId == null ) {

                // TODO: this will fail if we already "know" about the identifier

            	// FIXME: see https://redmine.dataone.org/issues/2572

                retPid = super.create(session, pid, object, sysmeta);

                result = (retPid.getValue().equals(pid.getValue()));

            }

        } catch (Exception e) {

            String msg = "Could not save object to local store (" + e.getClass().getName() + "): " + e.getMessage();

            failure = new ServiceFailure("2151", msg);

            setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.FAILED, failure);

            logMetacat.error(msg);

            throw new ServiceFailure("2151", msg);

        }

        // finish by setting the replication status

        setReplicationStatus(thisNodeSession, pid, nodeId, ReplicationStatus.COMPLETED, null);

        return result;

    }

    /**

     * Return the object identified by the given object identifier

     * 

     * @param session - the Session object containing the credentials for the Subject

     * @param pid - the object identifier for the given object

     * 

     * @return inputStream - the input stream of the given object

     * 

     * @throws InvalidToken

     * @throws ServiceFailure

     * @throws NotAuthorized

     * @throws InvalidRequest

     * @throws NotImplemented

     */

    @Override

    public InputStream get(Session session, Identifier pid) 

    throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented {

        return super.get(session, pid);

    }

    /**

     * Returns a Checksum for the specified object using an accepted hashing algorithm

     * 

     * @param session - the Session object containing the credentials for the Subject

     * @param pid - the object identifier for the given object

     * @param algorithm -  the name of an algorithm that will be used to compute 

     *                     a checksum of the bytes of the object

     * 

     * @return checksum - the checksum of the given object

     * 

     * @throws InvalidToken

     * @throws ServiceFailure

     * @throws NotAuthorized

     * @throws NotFound

     * @throws InvalidRequest

     * @throws NotImplemented

     */

    @Override

    public Checksum getChecksum(Session session, Identifier pid, String algorithm) 

        throws InvalidToken, ServiceFailure, NotAuthorized, NotFound,

        InvalidRequest, NotImplemented {

        Checksum checksum = null;

        String serviceFailure = "1410";

        String notFound = "1420";

        //Checkum only handles the pid, not sid

        checkV1SystemMetaPidExist(pid, serviceFailure, "The checksum for the object specified by "+pid.getValue()+" couldn't be returned ",  notFound, 

                "The object specified by "+pid.getValue()+" does not exist at this node.");

        InputStream inputStream = get(session, pid);

        try {

            checksum = ChecksumUtil.checksum(inputStream, algorithm);

        } catch (NoSuchAlgorithmException e) {

            throw new ServiceFailure("1410", "The checksum for the object specified by " + pid.getValue() + "could not be returned due to an internal error: "

                    + e.getMessage());

        } catch (IOException e) {

            throw new ServiceFailure("1410", "The checksum for the object specified by " + pid.getValue() + "could not be returned due to an internal error: "

                    + e.getMessage());

        }

        if (checksum == null) {

            throw new ServiceFailure("1410", "The checksum for the object specified by " + pid.getValue() + "could not be returned.");

        }

        return checksum;

    }

    /**

     * Return the system metadata for a given object

     * 

     * @param session - the Session object containing the credentials for the Subject

     * @param pid - the object identifier for the given object

     * 

     * @return inputStream - the input stream of the given system metadata object

     * 

     * @throws InvalidToken

     * @throws ServiceFailure

     * @throws NotAuthorized

     * @throws NotFound

     * @throws InvalidRequest

     * @throws NotImplemented

     */

    @Override

    public SystemMetadata getSystemMetadata(Session session, Identifier pid) 

        throws InvalidToken, ServiceFailure, NotAuthorized, NotFound,

        NotImplemented {

        return super.getSystemMetadata(session, pid);

    }

    /**