db_plugin.cpp

Go to the documentation of this file.

// =-=-=-=-=-=-=-
// irods includes
#include "rodsDef.h"
#include "authenticate.h"
#include "rodsQuota.h"
#include "msParam.h"
#include "rcConnect.h"
#include "icatStructs.hpp"
#include "icatHighLevelRoutines.hpp"
#include "mid_level.hpp"
#include "low_level.hpp"
 
// =-=-=-=-=-=-=-
// new irods includes
#include "irods_database_plugin.hpp"
#include "irods_database_constants.hpp"
#include "irods_postgres_object.hpp"
#include "irods_stacktrace.hpp"
#include "irods_catalog_properties.hpp"
#include "irods_sql_logger.hpp"
#include "irods_hierarchy_parser.hpp"
#include "irods_children_parser.hpp"
#include "irods_auth_object.hpp"
#include "irods_pam_auth_object.hpp"
#include "irods_auth_factory.hpp"
#include "irods_auth_plugin.hpp"
#include "irods_auth_manager.hpp"
#include "irods_auth_constants.hpp"
#include "irods_server_properties.hpp"
#include "irods_resource_manager.hpp"
#include "irods_virtual_path.hpp"
#include "irods_rs_comm_query.hpp"
#include "modAccessControl.h"
#include "checksum.hpp"
 
// =-=-=-=-=-=-=-
// irods includes
#include "rods.h"
#include "rcMisc.h"
#include "miscServerFunct.hpp"
 
// =-=-=-=-=-=-=-
// stl includes
#include <sstream>
#include <string>
#include <string_view>
#include <iostream>
#include <vector>
#include <boost/regex.hpp>
#include <boost/lexical_cast.hpp>
 
#include "irods_lexical_cast.hpp"
 
using leaf_bundle_t = irods::resource_manager::leaf_bundle_t;
extern irods::resource_manager resc_mgr;
 
extern int get64RandomBytes( char *buf );
extern int icatApplyRule( rsComm_t *rsComm, char *ruleName, char *arg1 );
 
static char prevChalSig[200]; /* a 'signature' of the previous
                          challenge.  This is used as a sessionSignature on the ICAT server
                          side.  Also see getSessionSignatureClientside function. */
 
 
//   Legal values for accessLevel in  chlModAccessControl (Access Parameter).
//   Defined here since other code does not need them (except for help messages)
#define AP_READ "read"
#define AP_WRITE "write"
#define AP_OWN "own"
#define AP_NULL "null"
 
static rodsLong_t MAX_PASSWORDS = 40;
/* TEMP_PASSWORD_TIME is the number of seconds the temporary, one-time
   password can be used.  chlCheckAuth also checks for this column
   to be < TEMP_PASSWORD_MAX_TIME (1000) to differentiate the row
   from regular passwords and PAM passwords.
   This time, 120 seconds, should be long enough to give the iDrop and
   iDrop-lite applets enough time to download and go through their
   startup sequence.  iDrop and iDrop-lite disconnect when idle to
   reduce the number of open connections and active agents.  */
 
#define PASSWORD_SCRAMBLE_PREFIX ".E_"
#define PASSWORD_KEY_ENV_VAR "IRODS_DATABASE_USER_PASSWORD_SALT"
#define PASSWORD_DEFAULT_KEY "a9_3fker"
 
#define MAX_HOST_STR 2700
 
// =-=-=-=-=-=-=-
// local variables externed for config file setting in
bool irods_pam_auth_no_extend = false;
size_t irods_pam_password_len = 20;
char irods_pam_password_min_time[ NAME_LEN ]     = { "121" };
char irods_pam_password_max_time[ NAME_LEN ]     = { "1209600" };
char irods_pam_password_default_time[ NAME_LEN ] = { "1209600" };
 
size_t log_sql_flg = 0;
icatSessionStruct icss; // JMC :: only for testing!!!
extern int logSQL;
 
int  creatingUserByGroupAdmin; // JMC - backport 4772
char mySessionTicket[NAME_LEN];
char mySessionClientAddr[NAME_LEN];
 
// =-=-=-=-=-=-=-
// property constants
const std::string ICSS_PROP( "irods_icss_property" );
const std::string ZONE_PROP( "irods_zone_property" );
 
// =-=-=-=-=-=-=-
// virtual path management
#define PATH_SEPARATOR irods::get_virtual_path_separator().c_str()
 
/*
   Parse the input fullUserNameIn into an output userName and userZone
   and check that the username is a valid format, meaning at most one
   '@' and at most one '#'.
   Full userNames are of the form user@department[#zone].
   It is assumed the output strings are at least NAME_LEN characters long
   and the input string is at most that long.
 */
int
validateAndParseUserName( const char *fullUserNameIn, char *userName, char *userZone ) {
    const std::string input( fullUserNameIn );
    boost::smatch matches;
    // This regex matches usernames with no hashes and optionally one at symbol,
    // and then optionally a hash followed by a zone name containing no hashes.
    //
    // Username must be between 1 and NAME_LEN-1 characters.
    // Username may contain any combination of word characters, @ symbols, dashes, and dots.
    // Username may not be . or .., as we create home directories for users
    const boost::regex expression( "((\\w|[-.@])+)(#([^#]*))?" );
    try {
        const bool matched = boost::regex_match( input, matches, expression );
        if ( !matched || matches.str( 1 ).size() >= NAME_LEN ||
                matches.str( 1 ).size() < 1 ||
                matches.str( 4 ).size() >= NAME_LEN ||
                matches.str( 1 ) == "." ||
                matches.str( 1 ) == ".." ) {
            if ( userName != NULL ) {
                userName[0] = '\0';
            }
            if ( userZone != NULL ) {
                userZone[0] = '\0';
            }
            return USER_INVALID_USERNAME_FORMAT;
        }
        if ( userName != NULL ) {
            snprintf( userName, NAME_LEN, "%s", matches.str( 1 ).c_str() );
        }
        if ( userZone != NULL ) {
            snprintf( userZone, NAME_LEN, "%s", matches.str( 4 ).c_str() );
        }
    }
    catch ( const boost::exception& ) {
        return SYS_INTERNAL_ERR;
    }
    return 0;
}
 
// =-=-=-=-=-=-=-
// helper fcn to handle cast to pg object
irods::error make_db_ptr(
    const irods::first_class_object_ptr& _fc,
    irods::postgres_object_ptr&          _pg ) {
    if ( !_fc.get() ) {
        return ERROR(
                   SYS_INVALID_INPUT_PARAM,
                   "incoming fco is null" );
 
    }
 
    _pg = boost::dynamic_pointer_cast <
          irods::postgres_object > (
              _fc );
 
    if ( _pg.get() ) {
        return SUCCESS();
 
    }
    else {
        return ERROR(
                   INVALID_DYNAMIC_CAST,
                   "failed to dynamic cast to postgres_object_ptr" );
    }
 
} // make_db_ptr
 
// =-=-=-=-=-=-=-
//  Called internally to rollback current transaction after an error.
int _rollback( const char *functionName ) {
    // =-=-=-=-=-=-=-
    // This type of rollback is needed for Postgres since the low-level
    // now does an automatic 'begin' to create a sql block */
    int status =  cmlExecuteNoAnswerSql( "rollback", &icss );
    if ( status == 0 ) {
        rodsLog( LOG_NOTICE,
                 "%s cmlExecuteNoAnswerSql(rollback) succeeded", functionName );
    }
    else {
        rodsLog( LOG_NOTICE,
                 "%s cmlExecuteNoAnswerSql(rollback) failure %d",
                 functionName, status );
    }
 
    return status;
 
} // _rollback
 
// =-=-=-=-=-=-=-
//  Internal function to return the local zone (which is the default
//  zone).  The first time it's called, it gets the zone from the DB and
//  subsequent calls just return that value.
irods::error getLocalZone(
    irods::plugin_property_map& _prop_map,
    icatSessionStruct*          _icss,
    std::string&                _zone ) {
    // =-=-=-=-=-=-=-
    // try to get the zone prop, if it is not cached
    // then we hit the catalog and request it
    irods::error ret = _prop_map.get< std::string >( ZONE_PROP, _zone );
    if ( !ret.ok() ) {
        char local_zone[ MAX_NAME_LEN ];
        int status;
        {
            std::vector<std::string> bindVars;
            bindVars.push_back( "local" );
            status = cmlGetStringValueFromSql(
                         ( char* )"select zone_name from R_ZONE_MAIN where zone_type_name=?",
                         local_zone, MAX_NAME_LEN, bindVars, _icss );
        }
        if ( status != 0 ) {
            _rollback( "getLocalZone" );
            return ERROR( status, "getLocalZone failure" );
        }
 
        // =-=-=-=-=-=-=-
        // set the zone property
        _zone = local_zone;
        ret = _prop_map.set< std::string >( ZONE_PROP, _zone );
        if ( !ret.ok() ) {
            return PASS( ret );
 
        }
 
    } // if no zone prop
 
    return SUCCESS();
 
} // getLocalZone
 
// =-=-=-=-=-=-=-
// @brief query for object found of a resource
int get_object_count_of_resource_by_name(
    icatSessionStruct* _icss,
    const std::string& _resc_name,
    rodsLong_t&         _count ) {
 
    rodsLong_t resc_id;
    irods::error ret = resc_mgr.hier_to_leaf_id(
                         _resc_name,
                         resc_id);
    if(!ret.ok()) {
        // if we have a bad resource in the database we need
        // to ignore this in order to still remove it
        if(SYS_RESC_DOES_NOT_EXIST == ret.code()) {
            _count = 0;
            return 0;
        }
        irods::log(PASS(ret));
        return ret.code();
    }
 
    std::string resc_id_str;
    ret = irods::lexical_cast<std::string>(resc_id, resc_id_str);
    if(!ret.ok()) {
        irods::log(PASS(ret));
        return ret.code();
    }
 
    std::vector<std::string> bindVars;
    bindVars.push_back( resc_id_str );
    int status = cmlGetIntegerValueFromSql(
                     ( char* )"select count(data_id) from R_DATA_MAIN where resc_id=?",
                     &_count,
                     bindVars,
                     _icss );
 
    return status;
 
} // get_object_count_of_resource_by_name
 
// =-=-=-=-=-=-=-
// @brief determine if user had write permission to data object
irods::error determine_user_has_modify_metadata_access(
    const std::string& _data_name,
    const std::string& _collection,
    const std::string& _user_name,
    const std::string& _zone ) {
 
    int status = 0;
 
    rodsLog(
        LOG_DEBUG,
        "%s :: [%s] [%s] [%s] [%s]",
        __FUNCTION__,
        _data_name.c_str(),
        _collection.c_str(),
        _user_name.c_str(),
        _zone.c_str() );
 
    // get the number of data object to which this will apply
    rodsLong_t num_data_objects = -1;
    {
        std::vector<std::string> bind_vars;
        bind_vars.push_back( _data_name );
        bind_vars.push_back( _collection );
        status = cmlGetIntegerValueFromSql(
                     "select count(DISTINCT DM.data_id) from R_DATA_MAIN DM, R_COLL_MAIN CM where DM.data_name like ? and DM.coll_id=CM.coll_id and CM.coll_name like ?",
                      &num_data_objects,
                      bind_vars,
                      &icss );
        if( 0 != status ) {
            _rollback( "chlAddAVUMetadataWild" );
            return ERROR(
                       status,
                       "failed to get object count" );
        }
 
        if( 0 == num_data_objects ) {
            std::string msg = "no data objects found for collection ";
            msg += _collection;
            msg += " and object name ";
            msg += _data_name;
            _rollback( "chlAddAVUMetadataWild" );
            return ERROR(
                       CAT_NO_ROWS_FOUND,
                       msg );
        }
    }
 
    // get the baseline 'access needed' value from the token table
    rodsLong_t access_needed = -1;
    {
        std::vector<std::string> bind_vars;
        int status = cmlGetIntegerValueFromSql(
                         "select token_id  from R_TOKN_MAIN where token_name = 'modify metadata' and token_namespace = 'access_type'",
             &access_needed,
                         bind_vars,
                         &icss );
        if( status < 0 ) {
            return ERROR(
                       status,
                       "query for modify metadata token_id failed" );
        }
    }
 
    // reproduce the creation of access permission entries
    // of "ACCESS_VIEW_ONE" and "ACCESS_VIEW_TWO"
    rodsLong_t access_permission = -1;
    {
#if ORA_ICAT
        std::string query = "select min(max_access_type_id) from ( select max(access_type_id) max_access_type_id from ( select access_type_id, DM.data_id from R_DATA_MAIN DM, R_OBJT_ACCESS OA, R_USER_GROUP UG, R_USER_MAIN UM, R_COLL_MAIN CM where DM.data_name like ? and DM.coll_id=CM.coll_id and CM.coll_name like ? and UM.user_name=? and UM.zone_name = ? and UM.user_type_name!='rodsgroup' and UM.user_id = UG.user_id and OA.object_id = DM.data_id and UG.group_user_id = OA.user_id ) group by data_id )";
#else
        std::string query = "select min(max_access_type_id) from ( select max(access_type_id) max_access_type_id from ( select access_type_id, DM.data_id from R_DATA_MAIN DM, R_OBJT_ACCESS OA, R_USER_GROUP UG, R_USER_MAIN UM, R_COLL_MAIN CM where DM.data_name like ? and DM.coll_id=CM.coll_id and CM.coll_name like ? and UM.user_name=? and UM.zone_name = ? and UM.user_type_name!='rodsgroup' and UM.user_id = UG.user_id and OA.object_id = DM.data_id and UG.group_user_id = OA.user_id ) as foo group by data_id ) as bar";
#endif
        std::vector<std::string> bind_vars;
        bind_vars.push_back( _data_name.c_str() );
        bind_vars.push_back( _collection.c_str() );
        bind_vars.push_back( _user_name.c_str() );
        bind_vars.push_back( _zone.c_str() );
        status = cmlGetIntegerValueFromSql(
                 query.c_str(),
                 &access_permission,
                     bind_vars,
                     &icss );
        if ( status == CAT_NO_ROWS_FOUND ) {
            _rollback( "chlAddAVUMetadataWild" );
            return ERROR(
                       CAT_NO_ACCESS_PERMISSION,
                       "access denied" );
        }
 
        if ( access_permission < access_needed ) {
            _rollback( "chlAddAVUMetadataWild" );
            return ERROR(
                       CAT_NO_ACCESS_PERMISSION,
                       "access denied" );
        }
    }
 
    // reproduce the count of access permission entries in "ACCESS_VIEW_TWO"
    rodsLong_t access_permission_count = -1;
    {
#if ORA_ICAT
        std::string query = "select count( max ) from ( select max(access_type_id) max from ( select access_type_id, DM.data_id from R_DATA_MAIN DM, R_OBJT_ACCESS OA, R_USER_GROUP UG, R_USER_MAIN UM, R_COLL_MAIN CM where DM.data_name like ? and DM.coll_id=CM.coll_id and CM.coll_name like ? and UM.user_name=? and UM.zone_name = ? and UM.user_type_name!='rodsgroup' and UM.user_id = UG.user_id and OA.object_id = DM.data_id and UG.group_user_id = OA.user_id ) group by data_id )";
#else
        std::string query = "select count( max ) from ( select max(access_type_id) max from ( select access_type_id, DM.data_id from R_DATA_MAIN DM, R_OBJT_ACCESS OA, R_USER_GROUP UG, R_USER_MAIN UM, R_COLL_MAIN CM where DM.data_name like ? and DM.coll_id=CM.coll_id and CM.coll_name like ? and UM.user_name=? and UM.zone_name = ? and UM.user_type_name!='rodsgroup' and UM.user_id = UG.user_id and OA.object_id = DM.data_id and UG.group_user_id = OA.user_id ) as foo group by data_id ) as baz";
#endif
        std::vector<std::string> bind_vars;
        bind_vars.push_back( _data_name.c_str() );
        bind_vars.push_back( _collection.c_str() );
        bind_vars.push_back( _user_name.c_str() );
        bind_vars.push_back( _zone.c_str() );
        status = cmlGetIntegerValueFromSql(
                 query.c_str(),
                 &access_permission_count,
                     bind_vars,
                     &icss );
        if( 0 != status ) {
            _rollback( "chlAddAVUMetadataWild" );
            return ERROR(
                       status,
                       "query for access permission count failed" );
        }
 
        if( num_data_objects > access_permission_count ) {
            std::stringstream msg;
            msg << "access denined - num_data_objects "
                << num_data_objects
                << " > access_permission_count "
                << access_permission_count;
            _rollback( "chlAddAVUMetadataWild" );
            return ERROR(
                       CAT_NO_ACCESS_PERMISSION,
                       msg.str() );
        }
    }
 
    // return number of data objects, keeping the semantics of the
    // original imeta addw operation
    return CODE( num_data_objects );
 
} // user_has_modify_metadata_access
 
 
/*
 * removeMetaMapAndAVU - remove AVU (user defined metadata) for an object,
 *   the metadata mapping information, if any.  Optionally, also remove
 *   any unused AVUs, if any, if some mapping information was removed.
 *
 */
void removeMetaMapAndAVU( char *dataObjNumber ) {
    char tSQL[MAX_SQL_SIZE];
    int status;
    cllBindVars[0] = dataObjNumber;
    cllBindVarCount = 1;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "removeMetaMapAndAVU SQL 1 " );
    }
    snprintf( tSQL, MAX_SQL_SIZE,
              "delete from R_OBJT_METAMAP where object_id=?" );
    status =  cmlExecuteNoAnswerSql( tSQL, &icss );
 
    /* Note, the status will be CAT_SUCCESS_BUT_WITH_NO_INFO (not 0) if
       there were no rows deleted from R_OBJT_METAMAP, in which case there
       is no need to do the SQL below.
    */
 
    return;
}
 
/*
 * removeAVUs - remove unused AVUs (user defined metadata), if any.
 */
static int removeAVUs() {
    char tSQL[MAX_SQL_SIZE];
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "removeAVUs SQL 1 " );
    }
    cllBindVarCount = 0;
 
#if ORA_ICAT
    snprintf( tSQL, MAX_SQL_SIZE,
              "delete from R_META_MAIN where meta_id in (select meta_id from R_META_MAIN minus select meta_id from R_OBJT_METAMAP)" );
#elif MY_ICAT
    /* MYSQL does not have 'minus' or 'except' (to my knowledge) so
     * use previous version of the SQL, which is very slow */
    snprintf( tSQL, MAX_SQL_SIZE,
              "delete from R_META_MAIN where meta_id not in (select meta_id from R_OBJT_METAMAP)" );
#else
    /* Postgres */
    snprintf( tSQL, MAX_SQL_SIZE,
              "delete from R_META_MAIN where meta_id in (select meta_id from R_META_MAIN except select meta_id from R_OBJT_METAMAP)" );
#endif
    const int status =  cmlExecuteNoAnswerSql( tSQL, &icss );
    rodsLog( LOG_DEBUG, "removeAVUs status=%d\n", status );
 
    return status;
}
 
int
_canConnectToCatalog(
    rsComm_t* _rsComm ) {
    int result = 0;
    if ( !icss.status ) {
        result = CATALOG_NOT_CONNECTED;
    }
    else if ( _rsComm->clientUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        result = CAT_INSUFFICIENT_PRIVILEGE_LEVEL;
    }
    else if ( _rsComm->proxyUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        result = CAT_INSUFFICIENT_PRIVILEGE_LEVEL;
    }
    return result;
}
 
static
int hostname_resolves_to_ipv4(const char* _hostname) {
    struct addrinfo hint;
    memset(&hint, 0, sizeof(hint));
    hint.ai_family = AF_INET;
    struct addrinfo *p_addrinfo;
    const int ret_getaddrinfo_with_retry = getaddrinfo_with_retry(_hostname, 0, &hint, &p_addrinfo);
    if (ret_getaddrinfo_with_retry) {
        return ret_getaddrinfo_with_retry;
    }
    freeaddrinfo(p_addrinfo);
    return 0;
}
 
 
int
_resolveHostName(rsComm_t* _rsComm, const char* _hostAddress) {
    const int status = hostname_resolves_to_ipv4(_hostAddress);
 
    if ( status != 0 ) {
        char errMsg[155];
        snprintf( errMsg, 150,
                  "Warning, resource host address '%s' is not a valid DNS entry, hostname_resolves_to_ipv4 failed.",
                  _hostAddress );
        addRErrorMsg( &_rsComm->rError, 0, errMsg );
    }
    if ( strcmp( _hostAddress, "localhost" ) == 0 ) {
        addRErrorMsg( &_rsComm->rError, 0,
                      "Warning, resource host address 'localhost' will not work properly as it maps to the local host from each client." );
    }
 
    return 0;
}
 
// Returns success if path is not root; otherwise, generates an error
irods::error
verify_non_root_vault_path(irods::plugin_context& _ctx, const std::string& path) {
    if (0 == path.compare("/")) {
        const std::string error_message = "root directory cannot be used as vault path.";
        addRErrorMsg(&_ctx.comm()->rError, 0, error_message.c_str());
        return ERROR(CAT_INVALID_RESOURCE_VAULT_PATH, error_message.c_str() );
    }
    return SUCCESS();
}
 
// =-=-=-=-=-=-=-
//
irods::error _childIsValid(
    irods::plugin_property_map& _prop_map,
    const std::string&          _new_child ) {
    // =-=-=-=-=-=-=-
    // Lookup the child resource and make sure its parent field is empty
    char parent[MAX_NAME_LEN];
    int status;
 
    // Get the resource name from the child string
    std::string resc_name;
    irods::children_parser parser;
    parser.set_string( _new_child );
    parser.first_child( resc_name );
 
    std::string zone;
    irods::error ret = getLocalZone( _prop_map, &icss, zone );
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // Get resource's parent
    irods::sql_logger logger( "_childIsValid", logSQL );
    logger.log();
    parent[0] = '\0';
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( resc_name );
        bindVars.push_back( zone );
        status = cmlGetStringValueFromSql(
                     "select resc_parent from R_RESC_MAIN where resc_name=? and zone_name=?",
                     parent, MAX_NAME_LEN, bindVars, &icss );
    }
    if ( status != 0 ) {
        if ( status == CAT_NO_ROWS_FOUND ) {
            std::stringstream ss;
            ss << "Child resource \"" << resc_name << "\" not found";
            irods::log( LOG_NOTICE, ss.str() );
            return ERROR( CHILD_NOT_FOUND, "child resource not found" );
        }
        else {
            _rollback( "_childIsValid" );
            return ERROR( status, "error encountered in query for _childIsValid" );
        }
    }
    else if ( strlen( parent ) != 0 ) {
        // If the resource already has a parent it cannot be added as a child of another one
        std::stringstream ss;
        ss << "Child resource \"" << resc_name << "\" already has a parent \"" << parent << "\"";
        irods::log( LOG_NOTICE, ss.str() );
        return ERROR( CHILD_HAS_PARENT, "child resource already has a parent" );
    }
    return SUCCESS();
}
 
irods::error _updateChildParent(
    const std::string& _child_resc_id,
    const std::string& _parent_resc_id,
    const std::string& _parent_child_context ) {
    irods::sql_logger logger( "_updateChildParent", logSQL );
 
    // Update the parent for the child resource
    // have to do this to get around const
    char myTime[50];
    getNowStr( myTime );
    cllBindVarCount = 0;
    cllBindVars[cllBindVarCount++] = _parent_resc_id.c_str();
    cllBindVars[cllBindVarCount++] = _parent_child_context.c_str();
    cllBindVars[cllBindVarCount++] = myTime;
    cllBindVars[cllBindVarCount++] = _child_resc_id.c_str();
    logger.log();
 
    int status = cmlExecuteNoAnswerSql(
                     "update R_RESC_MAIN set resc_parent=?, resc_parent_context=?, modify_ts=? "
                     "where resc_id=?",
                     &icss );
    if( status != 0 ) {
        _rollback( "_updateChildParent" );
        return ERROR( status, "cmlExecuteNoAnswerSql failed" );
    }
 
    return SUCCESS();
 
} // _updateChildParent
 
int
_rescHasData(
    icatSessionStruct* _icss,
    const std::string& _resc_name,
    bool&              _has_data ) {
    irods::sql_logger logger( "_rescHasData", logSQL );
    rodsLong_t obj_count{};
 
    logger.log();
 
    int status = get_object_count_of_resource_by_name(
                      _icss,
                      _resc_name,
                      obj_count );
    if( 0 == status ) {
        if ( 0 == obj_count ) {
            _has_data = false;
        }
    }
 
    return status;
}
 
irods::error validate_resource_name( std::string _resc_name ) {
 
    // Must be between 1 and NAME_LEN-1 characters.
    // Must start and end with a word character.
    // May contain non consecutive dashes.
    boost::regex re( "^(?=.{1,63}$)\\w+(-\\w+)*$" );
 
    if ( !boost::regex_match( _resc_name, re ) ) {
        std::stringstream msg;
        msg << "validate_resource_name failed for resource [";
        msg << _resc_name;
        msg << "]";
        return ERROR( SYS_INVALID_INPUT_PARAM, msg.str() );
    }
 
    return SUCCESS();
 
} // validate_resource_name
 
bool
_rescHasParentOrChild( char* rescId ) {
 
    char parent[MAX_NAME_LEN];
    char children[MAX_NAME_LEN];
    int status;
    irods::sql_logger logger( "_rescHasParentOrChild", logSQL );
 
    logger.log();
    parent[0] = '\0';
    children[0] = '\0';
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( rescId );
        status = cmlGetStringValueFromSql(
                     "select resc_parent from R_RESC_MAIN where resc_id=?",
                     parent, MAX_NAME_LEN, bindVars, &icss );
    }
    if ( status != 0 ) {
        if ( status == CAT_NO_ROWS_FOUND ) {
            std::stringstream ss;
            ss << "Resource \"" << rescId << "\" not found";
            irods::log( LOG_NOTICE, ss.str() );
        }
        else {
            _rollback( "_rescHasParentOrChild" );
        }
        return false;
    }
    if ( strlen( parent ) != 0 ) {
        return true;
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( rescId );
        status = cmlGetStringValueFromSql(
                     "select resc_id from R_RESC_MAIN where resc_parent=?",
                     children, MAX_NAME_LEN, bindVars, &icss );
    }
    if ( status != 0 ) {
        if ( status != CAT_NO_ROWS_FOUND ) {
            _rollback( "_rescHasParentOrChild" );
        }
        return false;
    }
    if ( strlen( children ) != 0 ) {
        return true;
    }
    return false;
 
}
 
bool _userInRUserAuth( const char* userName, const char* zoneName, const char* auth_name ) {
    int status;
    rodsLong_t iVal;
    irods::sql_logger logger( "_userInRUserAuth", logSQL );
 
    logger.log();
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( userName );
        bindVars.push_back( zoneName );
        bindVars.push_back( auth_name );
        status = cmlGetIntegerValueFromSql(
                    "select user_id from R_USER_AUTH where user_id=(select user_id from R_USER_MAIN where user_name=? and zone_name=?) and user_auth_name=?",
                    &iVal, bindVars, &icss );
    }
    if ( status != 0 ) {
        if ( status != CAT_NO_ROWS_FOUND ) {
            _rollback( "_userInRUserAuth" );
        }
        return false;
    } else {
        return true;
    }
}
 
// =-=-=-=-=-=-=-
static bool allowed_zone_char( const char _c ) {
    return ( !std::isalnum( _c ) &&
             !( '.' == _c )      &&
             !( '_' == _c ) );
} // allowed_zone_char
 
// =-=-=-=-=-=-=-
irods::error validate_zone_name(
    std::string _zone_name ) {
    std::string::iterator itr = std::find_if( _zone_name.begin(),
                                _zone_name.end(),
                                allowed_zone_char );
    if ( itr != _zone_name.end() || _zone_name.length() >= NAME_LEN ) {
        std::stringstream msg;
        msg << "validate_zone_name failed for zone [";
        msg << _zone_name;
        msg << "]";
        return ERROR( SYS_INVALID_INPUT_PARAM, msg.str() );
    }
 
    return SUCCESS();
 
} // validate_zone_name
 
/* delCollection (internally called),
   does not do the commit.
*/
static int _delColl( rsComm_t *rsComm, collInfo_t *collInfo ) {
    rodsLong_t iVal;
    char logicalEndName[MAX_NAME_LEN];
    char logicalParentDirName[MAX_NAME_LEN];
    char collIdNum[MAX_NAME_LEN];
    char parentCollIdNum[MAX_NAME_LEN];
    rodsLong_t status;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "_delColl" );
    }
 
    if ( !icss.status ) {
        return CATALOG_NOT_CONNECTED;
    }
 
    status = splitPathByKey( collInfo->collName,
                             logicalParentDirName, MAX_NAME_LEN, logicalEndName, MAX_NAME_LEN, '/' );
 
    if ( strlen( logicalParentDirName ) == 0 ) {
        snprintf( logicalParentDirName, sizeof( logicalParentDirName ), "%s", PATH_SEPARATOR );
        snprintf( logicalEndName, sizeof( logicalEndName ), "%s", collInfo->collName + 1 );
    }
 
    /* Check that the parent collection exists and user has write permission,
       and get the collectionID */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "_delColl SQL 1 " );
    }
    status = cmlCheckDir( logicalParentDirName,
                          rsComm->clientUser.userName,
                          rsComm->clientUser.rodsZone,
                          ACCESS_MODIFY_OBJECT,
                          &icss );
    if ( status < 0 ) {
        char errMsg[105];
        if ( status == CAT_UNKNOWN_COLLECTION ) {
            snprintf( errMsg, 100, "collection '%s' is unknown",
                      logicalParentDirName );
            addRErrorMsg( &rsComm->rError, 0, errMsg );
            return status;
        }
        _rollback( "_delColl" );
        return status;
    }
    snprintf( parentCollIdNum, MAX_NAME_LEN, "%lld", status );
 
    /* Check that the collection exists and user has DELETE or better
       permission */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "_delColl SQL 2" );
    }
    status = cmlCheckDir( collInfo->collName,
                          rsComm->clientUser.userName,
                          rsComm->clientUser.rodsZone,
                          ACCESS_DELETE_OBJECT,
                          &icss );
    if ( status < 0 ) {
        return status;
    }
    snprintf( collIdNum, MAX_NAME_LEN, "%lld", status );
 
    /* check that the collection is empty (both subdirs and files) */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "_delColl SQL 3" );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( collInfo->collName );
        bindVars.push_back( collInfo->collName );
        status = cmlGetIntegerValueFromSql(
                     "select coll_id from R_COLL_MAIN where parent_coll_name=? union select coll_id from R_DATA_MAIN where coll_id=(select coll_id from R_COLL_MAIN where coll_name=?)",
                     &iVal, bindVars, &icss );
    }
    if ( status != CAT_NO_ROWS_FOUND ) {
        return CAT_COLLECTION_NOT_EMPTY;
    }
 
    /* delete the row if it exists */
    /* The use of coll_id isn't really needed but may add a little safety.
       Previously, we included a check that it was owned by the user but
       the above cmlCheckDir is more accurate (handles group access). */
    cllBindVars[cllBindVarCount++] = collInfo->collName;
    cllBindVars[cllBindVarCount++] = collIdNum;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "_delColl SQL 4" );
    }
    status =  cmlExecuteNoAnswerSql(
                  "delete from R_COLL_MAIN where coll_name=? and coll_id=?",
                  &icss );
    if ( status != 0 ) { /* error, odd one as everything checked above */
        rodsLog( LOG_NOTICE,
                 "_delColl cmlExecuteNoAnswerSql delete failure %d",
                 status );
        _rollback( "_delColl" );
        return status;
    }
 
    /* remove any access rows */
    cllBindVars[cllBindVarCount++] = collIdNum;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "_delColl SQL 5" );
    }
    status =  cmlExecuteNoAnswerSql(
                  "delete from R_OBJT_ACCESS where object_id=?",
                  &icss );
    if ( status != 0 ) { /* error, odd one as everything checked above */
        rodsLog( LOG_NOTICE,
                 "_delColl cmlExecuteNoAnswerSql delete access failure %d",
                 status );
        _rollback( "_delColl" );
    }
 
    /* Remove associated AVUs, if any */
    removeMetaMapAndAVU( collIdNum );
 
    /* Audit */
    status = cmlAudit3( AU_DELETE_COLL,
                        collIdNum,
                        rsComm->clientUser.userName,
                        rsComm->clientUser.rodsZone,
                        collInfo->collName,
                        &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlModColl cmlAudit3 failure %d",
                 status );
        _rollback( "_delColl" );
        return status;
    }
 
    return status;
 
} // _delColl
 
// =-=-=-=-=-=-=-
// local function to delegate the response
// verification to an authentication plugin
static
irods::error verify_auth_response(
    const char* _scheme,
    const char* _challenge,
    const char* _user_name,
    const char* _response ) {
    // =-=-=-=-=-=-=-
    // validate incoming parameters
    if ( !_scheme ) {
        return ERROR( SYS_INVALID_INPUT_PARAM, "null _scheme ptr" );
    }
    else if ( !_challenge ) {
        return ERROR( SYS_INVALID_INPUT_PARAM, "null _challenge ptr" );
    }
    else if ( !_user_name ) {
        return ERROR( SYS_INVALID_INPUT_PARAM, "null _user_name ptr" );
    }
    else if ( !_response ) {
        return ERROR( SYS_INVALID_INPUT_PARAM, "null _response ptr" );
    }
 
    // =-=-=-=-=-=-=-
    // construct an auth object given the scheme
    irods::auth_object_ptr auth_obj;
    irods::error ret = irods::auth_factory( _scheme, 0, auth_obj );
    if ( !ret.ok() ) {
        return ret;
    }
 
    // =-=-=-=-=-=-=-
    // resolve an auth plugin given the auth object
    irods::plugin_ptr ptr;
    ret = auth_obj->resolve( irods::AUTH_INTERFACE, ptr );
    if ( !ret.ok() ) {
        return ret;
    }
    irods::auth_ptr auth_plugin = boost::dynamic_pointer_cast< irods::auth >( ptr );
 
    // =-=-=-=-=-=-=-
    // call auth verify on plugin
    ret = auth_plugin->call <const char*, const char*, const char* > ( 0, irods::AUTH_AGENT_AUTH_VERIFY, auth_obj, _challenge, _user_name, _response );
    if ( !ret.ok() ) {
        irods::log( PASS( ret ) );
        return ret;
    }
 
    return SUCCESS();
 
} // verify_auth_response
 
/*
   Possibly descramble a password (for user passwords stored in the ICAT).
   Called internally, from various chl functions.
*/
static int
icatDescramble( char *pw ) {
    char *cp1, *cp2, *cp3;
    int i, len;
    char pw2[MAX_PASSWORD_LEN + 10];
    char unscrambled[MAX_PASSWORD_LEN + 10];
 
    len = strlen( PASSWORD_SCRAMBLE_PREFIX );
    cp1 = pw;
    cp2 = PASSWORD_SCRAMBLE_PREFIX; /* if starts with this, it is scrambled */
    for ( i = 0; i < len; i++ ) {
        if ( *cp1++ != *cp2++ ) {
            return 0;                /* not scrambled, leave as is */
        }
    }
    snprintf( pw2, sizeof( pw2 ), "%s", cp1 );
    cp3 = getenv( PASSWORD_KEY_ENV_VAR );
    if ( cp3 == NULL ) {
        cp3 = PASSWORD_DEFAULT_KEY;
    }
    obfDecodeByKey( pw2, cp3, unscrambled );
    strncpy( pw, unscrambled, MAX_PASSWORD_LEN );
 
    return 0;
}
 
/*
   Scramble a password (for user passwords stored in the ICAT).
   Called internally.
*/
static int
icatScramble( char *pw ) {
    char *cp1;
    char newPw[MAX_PASSWORD_LEN + 10];
    char scrambled[MAX_PASSWORD_LEN + 10];
 
    cp1 = getenv( PASSWORD_KEY_ENV_VAR );
    if ( cp1 == NULL ) {
        cp1 = PASSWORD_DEFAULT_KEY;
    }
    obfEncodeByKey( pw, cp1, scrambled );
    snprintf( newPw, sizeof( newPw ), "%s%s", PASSWORD_SCRAMBLE_PREFIX, scrambled );
    strncpy( pw, newPw, MAX_PASSWORD_LEN );
    return 0;
}
 
/*
  de-scramble a password sent from the client.
  This isn't real encryption, but does obfuscate the pw on the network.
  Called internally, from chlModUser.
*/
int decodePw( rsComm_t *rsComm, const char *in, char *out ) {
    int status;
    char *cp;
    char password[MAX_PASSWORD_LEN];
    char upassword[MAX_PASSWORD_LEN + 10];
    char rand[] =
        "1gCBizHWbwIYyWLo";  /* must match clients */
    int pwLen1, pwLen2;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "decodePw - SQL 1 " );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( rsComm->clientUser.userName );
        bindVars.push_back( rsComm->clientUser.rodsZone );
        status = cmlGetStringValueFromSql(
                     "select rcat_password from R_USER_PASSWORD, R_USER_MAIN where user_name=? and R_USER_MAIN.zone_name=? and R_USER_MAIN.user_id = R_USER_PASSWORD.user_id",
                     password, MAX_PASSWORD_LEN, bindVars, &icss );
    }
    if ( status != 0 ) {
        if ( status == CAT_NO_ROWS_FOUND ) {
            status = CAT_INVALID_USER; /* Be a little more specific */
        }
        else {
            _rollback( "decodePw" );
        }
        return status;
    }
 
    icatDescramble( password );
 
    obfDecodeByKeyV2( in, password, prevChalSig, upassword );
 
    pwLen1 = strlen( upassword );
 
    memset( password, 0, MAX_PASSWORD_LEN );
 
    cp = strstr( upassword, rand );
    if ( cp != NULL ) {
        *cp = '\0';
    }
 
    pwLen2 = strlen( upassword );
 
    if ( pwLen2 > MAX_PASSWORD_LEN - 5 && pwLen2 == pwLen1 ) {
        /* probable failure */
        char errMsg[260];
        snprintf( errMsg, 250,
                  "Error with password encoding.  This can be caused by not connecting directly to the ICAT host, not using password authentication (using GSI or Kerberos instead), or entering your password incorrectly (if prompted)." );
        addRErrorMsg( &rsComm->rError, 0, errMsg );
        return CAT_PASSWORD_ENCODING_ERROR;
    }
    strcpy( out, upassword );
    memset( upassword, 0, MAX_PASSWORD_LEN );
 
    return 0;
}
 
int
convertTypeOption( const char *typeStr ) {
    if ( strcmp( typeStr, "-d" ) == 0 ) {
        return ( 1 );   /* dataObj */
    }
    if ( strcmp( typeStr, "-D" ) == 0 ) {
        return ( 1 );   /* dataObj */
    }
    if ( strcmp( typeStr, "-c" ) == 0 ) {
        return ( 2 );   /* collection */
    }
    if ( strcmp( typeStr, "-C" ) == 0 ) {
        return ( 2 );   /* collection */
    }
    if ( strcmp( typeStr, "-r" ) == 0 ) {
        return ( 3 );   /* resource */
    }
    if ( strcmp( typeStr, "-R" ) == 0 ) {
        return ( 3 );   /* resource */
    }
    if ( strcmp( typeStr, "-u" ) == 0 ) {
        return ( 4 );   /* user */
    }
    if ( strcmp( typeStr, "-U" ) == 0 ) {
        return ( 4 );   /* user */
    }
    return 0;
}
 
/*
  Check object - get an object's ID and check that the user has access.
  Called internally.
*/
rodsLong_t checkAndGetObjectId(
    rsComm_t*                   rsComm,
    irods::plugin_property_map& _prop_map,
    const char*                 type,
    const char*                 name,
    const char*                 access ) {
    int itype;
    char logicalEndName[MAX_NAME_LEN];
    char logicalParentDirName[MAX_NAME_LEN];
    rodsLong_t status;
    rodsLong_t objId;
    char userName[NAME_LEN];
    char userZone[NAME_LEN];
 
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "checkAndGetObjectId" );
    }
 
    if ( !icss.status ) {
        return CATALOG_NOT_CONNECTED;
    }
 
    if ( type == NULL ) {
        return CAT_INVALID_ARGUMENT;
    }
 
    if ( *type == '\0' ) {
        return CAT_INVALID_ARGUMENT;
    }
 
 
    if ( name == NULL ) {
        return CAT_INVALID_ARGUMENT;
    }
 
    if ( *name == '\0' ) {
        return CAT_INVALID_ARGUMENT;
    }
 
 
    itype = convertTypeOption( type );
    if ( itype == 0 ) {
        return CAT_INVALID_ARGUMENT;
    }
 
    if ( itype == 1 ) {
        status = splitPathByKey( name,
                                 logicalParentDirName, MAX_NAME_LEN, logicalEndName, MAX_NAME_LEN, '/' );
        if ( strlen( logicalParentDirName ) == 0 ) {
            snprintf( logicalParentDirName, sizeof( logicalParentDirName ), "%s", PATH_SEPARATOR );
            snprintf( logicalEndName, sizeof( logicalEndName ), "%s", name );
        }
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "checkAndGetObjectId SQL 1 " );
        }
        status = cmlCheckDataObjOnly( logicalParentDirName, logicalEndName,
                                      rsComm->clientUser.userName,
                                      rsComm->clientUser.rodsZone,
                                      access, &icss );
        if ( status < 0 ) {
            _rollback( "checkAndGetObjectId" );
            return status;
        }
        objId = status;
    }
 
    if ( itype == 2 ) {
        /* Check that the collection exists and user has create_metadata permission,
           and get the collectionID */
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "checkAndGetObjectId SQL 2" );
        }
        status = cmlCheckDir( name,
                              rsComm->clientUser.userName,
                              rsComm->clientUser.rodsZone,
                              access, &icss );
        if ( status < 0 ) {
            char errMsg[105];
            if ( status == CAT_UNKNOWN_COLLECTION ) {
                snprintf( errMsg, 100, "collection '%s' is unknown",
                          name );
                addRErrorMsg( &rsComm->rError, 0, errMsg );
            }
            return status;
        }
        objId = status;
    }
 
    if ( itype == 3 ) {
        if ( rsComm->clientUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
            return CAT_INSUFFICIENT_PRIVILEGE_LEVEL;
        }
 
        std::string zone;
        irods::error ret = getLocalZone( _prop_map, &icss, zone );
        if ( !ret.ok() ) {
            return PASS( ret ).code();
        }
 
        objId = 0;
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "checkAndGetObjectId SQL 3" );
        }
        {
            std::vector<std::string> bindVars;
            bindVars.push_back( name );
            bindVars.push_back( zone );
            status = cmlGetIntegerValueFromSql(
                         "select resc_id from R_RESC_MAIN where resc_name=? and zone_name=?",
                         &objId, bindVars, &icss );
        }
        if ( status != 0 ) {
            if ( status == CAT_NO_ROWS_FOUND ) {
                return CAT_INVALID_RESOURCE;
            }
            _rollback( "checkAndGetObjectId" );
            return status;
        }
    }
 
    if ( itype == 4 ) {
        if ( rsComm->clientUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
            return CAT_INSUFFICIENT_PRIVILEGE_LEVEL;
        }
 
        status = validateAndParseUserName( name, userName, userZone );
        if ( status ) {
            return status;
        }
        if ( userZone[0] == '\0' ) {
            std::string zone;
            irods::error ret = getLocalZone( _prop_map, &icss, zone );
            if ( !ret.ok() ) {
                return PASS( ret ).code();
            }
            snprintf( userZone, sizeof( userZone ), "%s",  zone.c_str() );
        }
 
        objId = 0;
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "checkAndGetObjectId SQL 4" );
        }
        {
            std::vector<std::string> bindVars;
            bindVars.push_back( userName );
            bindVars.push_back( userZone );
            status = cmlGetIntegerValueFromSql(
                         "select user_id from R_USER_MAIN where user_name=? and zone_name=?",
                         &objId, bindVars, &icss );
        }
        if ( status != 0 ) {
            if ( status == CAT_NO_ROWS_FOUND ) {
                return CAT_INVALID_USER;
            }
            _rollback( "checkAndGetObjectId" );
            return status;
        }
    }
 
    return objId;
}
 
/*
// =-=-=-=-=-=-=-
// JMC - backport 4836
+ Find existing AVU triplet.
+ Return code is error or the AVU ID.
+*/
rodsLong_t
findAVU( const char *attribute, const char *value, const char *units ) {
    rodsLong_t status;
// =-=-=-=-=-=-=-
 
    rodsLong_t iVal;
    iVal = 0;
    if ( *units != '\0' ) {
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "findAVU SQL 1" );    // JMC - backport 4836
        }
        {
            std::vector<std::string> bindVars;
            bindVars.push_back( attribute );
            bindVars.push_back( value );
            bindVars.push_back( units );
            status = cmlGetIntegerValueFromSql(
                         "select meta_id from R_META_MAIN where meta_attr_name=? and meta_attr_value=? and meta_attr_unit=?",
                         &iVal, bindVars, &icss );
        }
    }
    else {
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "findAVU SQL 2" );
        }
        {
            std::vector<std::string> bindVars;
            bindVars.push_back( attribute );
            bindVars.push_back( value );
            status = cmlGetIntegerValueFromSql(
                         "select meta_id from R_META_MAIN where meta_attr_name=? and meta_attr_value=? and (meta_attr_unit='' or meta_attr_unit IS NULL)", // JMC - backport 4827
                         &iVal, bindVars, &icss );
        }
    }
    if ( status == 0 ) {
        status = iVal; /* use existing R_META_MAIN row */
        return status;
    }
// =-=-=-=-=-=-=-
// JMC - backport 4836
    return ( status ); // JMC - backport 4836
}
 
/*
  Find existing or insert a new AVU triplet.
  Return code is error, or the AVU ID.
*/
int
findOrInsertAVU( const char *attribute, const char *value, const char *units ) {
    char nextStr[MAX_NAME_LEN];
    char myTime[50];
    rodsLong_t status, seqNum;
    rodsLong_t iVal;
    iVal = findAVU( attribute, value, units );
    if ( iVal > 0 ) {
        return iVal;
    }
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "findOrInsertAVU SQL 1" );
    }
// =-=-=-=-=-=-=-
    status = cmlGetNextSeqVal( &icss );
    if ( status < 0 ) {
        rodsLog( LOG_NOTICE, "findOrInsertAVU cmlGetNextSeqVal failure %d",
                 status );
        return status;
    }
    seqNum = status; /* the returned status is the next sequence value */
 
    snprintf( nextStr, sizeof nextStr, "%lld", seqNum );
 
    getNowStr( myTime );
 
    cllBindVars[cllBindVarCount++] = nextStr;
    cllBindVars[cllBindVarCount++] = attribute;
    cllBindVars[cllBindVarCount++] = value;
    cllBindVars[cllBindVarCount++] = units;
    cllBindVars[cllBindVarCount++] = myTime;
    cllBindVars[cllBindVarCount++] = myTime;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "findOrInsertAVU SQL 2" );    // JMC - backport 4836
    }
    status =  cmlExecuteNoAnswerSql(
                  "insert into R_META_MAIN (meta_id, meta_attr_name, meta_attr_value, meta_attr_unit, create_ts, modify_ts) values (?, ?, ?, ?, ?, ?)",
                  &icss );
    if ( status < 0 ) {
        rodsLog( LOG_NOTICE, "findOrInsertAVU insert failure %d", status );
        return status;
    }
    return seqNum;
}
 
 
/* create a path name with escaped SQL special characters (% and _) */
std::string
makeEscapedPath( const std::string &inPath ) {
    return boost::regex_replace( inPath, boost::regex( "[%_\\\\]" ), "\\\\$&" );
}
 
/* Internal routine to modify inheritance */
/* inheritFlag =1 to set, 2 to remove */
int _modInheritance( int inheritFlag, int recursiveFlag, const char *collIdStr, const char *pathName ) {
 
    const char* newValue = inheritFlag == 1 ? "1" : "0";
 
    char myTime[50];
    getNowStr( myTime );
 
    rodsLong_t status;
    /* non-Recursive mode */
    if ( recursiveFlag == 0 ) {
 
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "_modInheritance SQL 1" );
        }
 
        cllBindVars[cllBindVarCount++] = newValue;
        cllBindVars[cllBindVarCount++] = myTime;
        cllBindVars[cllBindVarCount++] = collIdStr;
        status =  cmlExecuteNoAnswerSql(
                      "update R_COLL_MAIN set coll_inheritance=?, modify_ts=? where coll_id=?",
                      &icss );
    }
    else {
        /* Recursive mode */
        std::string pathStart = makeEscapedPath( pathName ) + "/%";
 
        cllBindVars[cllBindVarCount++] = newValue;
        cllBindVars[cllBindVarCount++] = myTime;
        cllBindVars[cllBindVarCount++] = pathName;
        cllBindVars[cllBindVarCount++] = pathStart.c_str();
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "_modInheritance SQL 2" );
        }
#ifdef ORA_ICAT
        status =  cmlExecuteNoAnswerSql(
                      "update R_COLL_MAIN set coll_inheritance=?, modify_ts=? where coll_name = ? or coll_name like ? ESCAPE '\\'",
                      &icss );
#else
        status =  cmlExecuteNoAnswerSql(
                      "update R_COLL_MAIN set coll_inheritance=?, modify_ts=? where coll_name = ? or coll_name like ?",
                      &icss );
#endif
    }
    if ( status != 0 ) {
        _rollback( "_modInheritance" );
        return status;
    }
 
    char auditStr[30];
    snprintf( auditStr, sizeof( auditStr ), "inheritance %srecursive %s",
              recursiveFlag ? "" : "non-",
              newValue );
 
    /* Audit */
    status = cmlAudit5( AU_MOD_ACCESS_CONTROL_COLL,
                        collIdStr,
                        "0",
                        auditStr,
                        &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "_modInheritance cmlAudit5 failure %d",
                 status );
        _rollback( "_modInheritance" );
        return status;
    }
 
    status =  cmlExecuteNoAnswerSql( "commit", &icss );
    return status;
}
 
/*
  Set the over_quota values (if any) using the limits and
  and the current usage; handling the various types: per-user per-resource,
  per-user total-usage, group per-resource, and group-total.
 
  The over_quota column is positive if over_quota and the negative value
  indicates how much space is left before reaching the quota.
*/
int setOverQuota( rsComm_t *rsComm ) {
    int status;
    int rowsFound;
    int statementNum = UNINITIALIZED_STATEMENT_NUMBER;
    char myTime[50];
 
    /* For each defined group limit (if any), get a total usage on that
     * resource for all users in that group: */
    char mySQL1[] = "select sum(quota_usage), UM1.user_id, R_QUOTA_USAGE.resc_id from R_QUOTA_USAGE, R_QUOTA_MAIN, R_USER_MAIN UM1, R_USER_GROUP, R_USER_MAIN UM2 where R_QUOTA_MAIN.user_id = UM1.user_id and UM1.user_type_name = 'rodsgroup' and R_USER_GROUP.group_user_id = UM1.user_id and UM2.user_id = R_USER_GROUP.user_id and R_QUOTA_USAGE.user_id = UM2.user_id and R_QUOTA_MAIN.resc_id = R_QUOTA_USAGE.resc_id group by UM1.user_id, R_QUOTA_USAGE.resc_id";
 
    /* For each defined group limit on total usage (if any), get a
     * total usage on any resource for all users in that group: */
    char mySQL2a[] = "select sum(quota_usage), R_QUOTA_MAIN.quota_limit, UM1.user_id from R_QUOTA_USAGE, R_QUOTA_MAIN, R_USER_MAIN UM1, R_USER_GROUP, R_USER_MAIN UM2 where R_QUOTA_MAIN.user_id = UM1.user_id and UM1.user_type_name = 'rodsgroup' and R_USER_GROUP.group_user_id = UM1.user_id and UM2.user_id = R_USER_GROUP.user_id and R_QUOTA_USAGE.user_id = UM2.user_id and R_QUOTA_USAGE.resc_id != %s and R_QUOTA_MAIN.resc_id = %s group by UM1.user_id,  R_QUOTA_MAIN.quota_limit";
    char mySQL2b[MAX_SQL_SIZE];
 
    char mySQL3a[] = "update R_QUOTA_MAIN set quota_over= %s - ?, modify_ts=? where user_id=? and %s - ? > quota_over";
    char mySQL3b[MAX_SQL_SIZE];
 
 
    /* Initialize over_quota values (if any) to the no-usage value
       which is the negative of the limit.  */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "setOverQuota SQL 1" );
    }
    status =  cmlExecuteNoAnswerSql(
                  "update R_QUOTA_MAIN set quota_over = -quota_limit", &icss );
    if ( status == CAT_SUCCESS_BUT_WITH_NO_INFO ) {
        return ( 0 );   /* no quotas, done */
    }
    if ( status != 0 ) {
        return status;
    }
 
    /* Set the over_quota values for per-resource, if any */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "setOverQuota SQL 2" );
    }
    status =  cmlExecuteNoAnswerSql(
#if ORA_ICAT
                  "update R_QUOTA_MAIN set quota_over = (select distinct R_QUOTA_USAGE.quota_usage - R_QUOTA_MAIN.quota_limit from R_QUOTA_USAGE where R_QUOTA_MAIN.user_id = R_QUOTA_USAGE.user_id and R_QUOTA_MAIN.resc_id = R_QUOTA_USAGE.resc_id) where exists (select 1 from R_QUOTA_USAGE where R_QUOTA_MAIN.user_id = R_QUOTA_USAGE.user_id and R_QUOTA_MAIN.resc_id = R_QUOTA_USAGE.resc_id)",
#elif MY_ICAT
                  "update R_QUOTA_MAIN, R_QUOTA_USAGE set R_QUOTA_MAIN.quota_over = R_QUOTA_USAGE.quota_usage - R_QUOTA_MAIN.quota_limit where R_QUOTA_MAIN.user_id = R_QUOTA_USAGE.user_id and R_QUOTA_MAIN.resc_id = R_QUOTA_USAGE.resc_id",
#else
                  "update R_QUOTA_MAIN set quota_over = quota_usage - quota_limit from R_QUOTA_USAGE where R_QUOTA_MAIN.user_id = R_QUOTA_USAGE.user_id and R_QUOTA_MAIN.resc_id = R_QUOTA_USAGE.resc_id",
#endif
                  & icss );
    if ( status == CAT_SUCCESS_BUT_WITH_NO_INFO ) {
        status = 0;    /* none */
    }
    if ( status != 0 ) {
        return status;
    }
 
    /* Set the over_quota values for irods-total, if any, and only if
       the this over_quota value is higher than the previous.  Do it in
       two steps to keep it simplier (there may be a better way tho).
    */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "setOverQuota SQL 3" );
    }
    getNowStr( myTime );
    for ( rowsFound = 0;; rowsFound++ ) {
        int status2;
        if ( rowsFound == 0 ) {
            status = cmlGetFirstRowFromSql( "select sum(quota_usage), R_QUOTA_MAIN.user_id from R_QUOTA_USAGE, R_QUOTA_MAIN where R_QUOTA_MAIN.user_id = R_QUOTA_USAGE.user_id and R_QUOTA_MAIN.resc_id = '0' group by R_QUOTA_MAIN.user_id",
                                            &statementNum, 0, &icss );
        }
        else {
            status = cmlGetNextRowFromStatement( statementNum, &icss );
        }
        if ( status != 0 ) {
            break;
        }
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[0];
        cllBindVars[cllBindVarCount++] = myTime;
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[1];
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[0];
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "setOverQuota SQL 4" );
        }
        status2 = cmlExecuteNoAnswerSql( "update R_QUOTA_MAIN set quota_over=?-quota_limit, modify_ts=? where user_id=? and ?-quota_limit > quota_over and resc_id='0'",
                                         &icss );
        if ( status2 == CAT_SUCCESS_BUT_WITH_NO_INFO ) {
            status2 = 0;
        }
        if ( status2 != 0 ) {
            return status2;
        }
    }
 
    cmlFreeStatement(statementNum, &icss);
 
    /* Handle group quotas on resources */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "setOverQuota SQL 5" );
    }
    for ( rowsFound = 0;; rowsFound++ ) {
        int status2;
        if ( rowsFound == 0 ) {
            status = cmlGetFirstRowFromSql( mySQL1, &statementNum,
                                            0, &icss );
        }
        else {
            status = cmlGetNextRowFromStatement( statementNum, &icss );
        }
        if ( status != 0 ) {
            break;
        }
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[0];
        cllBindVars[cllBindVarCount++] = myTime;
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[1];
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[0];
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[2];
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "setOverQuota SQL 6" );
        }
        status2 = cmlExecuteNoAnswerSql( "update R_QUOTA_MAIN set quota_over=?-quota_limit, modify_ts=? where user_id=? and ?-quota_limit > quota_over and R_QUOTA_MAIN.resc_id=?",
                                         &icss );
        if ( status2 == CAT_SUCCESS_BUT_WITH_NO_INFO ) {
            status2 = 0;
        }
        if ( status2 != 0 ) {
            cmlFreeStatement(statementNum, &icss);
            return status2;
        }
    }
    if ( status == CAT_NO_ROWS_FOUND ) {
        status = 0;
    }
    if ( status != 0 ) {
        cmlFreeStatement(statementNum, &icss);
        return status;
    }
 
    cmlFreeStatement(statementNum, &icss);
 
    /* Handle group quotas on total usage */
#if ORA_ICAT
    /* For Oracle cast is to integer, for Postgres to bigint,for MySQL no cast*/
    snprintf( mySQL2b, sizeof mySQL2b, mySQL2a,
              "cast('0' as integer)", "cast('0' as integer)" );
    snprintf( mySQL3b, sizeof mySQL3b, mySQL3a,
              "cast(? as integer)", "cast(? as integer)" );
#elif MY_ICAT
    snprintf( mySQL2b, sizeof mySQL2b, mySQL2a, "'0'", "'0'" );
    snprintf( mySQL3b, sizeof mySQL3b, mySQL3a, "?", "?" );
#else
    snprintf( mySQL2b, sizeof mySQL2b, mySQL2a,
              "cast('0' as bigint)", "cast('0' as bigint)" );
    snprintf( mySQL3b, sizeof mySQL3b, mySQL3a,
              "cast(? as bigint)", "cast(? as bigint)" );
#endif
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "setOverQuota SQL 7" );
    }
    getNowStr( myTime );
    for ( rowsFound = 0;; rowsFound++ ) {
        int status2;
        if ( rowsFound == 0 ) {
            status = cmlGetFirstRowFromSql( mySQL2b, &statementNum,
                                            0, &icss );
        }
        else {
            status = cmlGetNextRowFromStatement( statementNum, &icss );
        }
        if ( status != 0 ) {
            break;
        }
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[0];
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[1];
        cllBindVars[cllBindVarCount++] = myTime;
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[2];
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[0];
        cllBindVars[cllBindVarCount++] = icss.stmtPtr[statementNum]->resultValue[1];
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "setOverQuota SQL 8" );
        }
        status2 = cmlExecuteNoAnswerSql( mySQL3b,
                                         &icss );
        if ( status2 == CAT_SUCCESS_BUT_WITH_NO_INFO ) {
            status2 = 0;
        }
        if ( status2 != 0 ) {
            cmlFreeStatement(statementNum, &icss);
            return status2;
        }
    }
    if ( status == CAT_NO_ROWS_FOUND ) {
        status = 0;
    }
    if ( status != 0 ) {
        cmlFreeStatement(statementNum, &icss);
        return status;
    }
 
    /* To simplify the query, if either of the above group operations
       found some over_quota, will probably want to update and insert rows
       for each user into R_QUOTA_MAIN.  For now tho, this is not done and
       perhaps shouldn't be, to keep it a little less complicated. */
 
    cmlFreeStatement(statementNum, &icss);
    return status;
}
 
int
icatGetTicketUserId( irods::plugin_property_map& _prop_map, const char *userName, char *userIdStr ) {
 
    char userId[NAME_LEN];
    char userZone[NAME_LEN];
    char zoneToUse[NAME_LEN];
    char userName2[NAME_LEN];
    int status;
 
    std::string zone;
    irods::error ret = getLocalZone( _prop_map, &icss, zone );
    if ( !ret.ok() ) {
        return ret.code();
    }
 
    snprintf( zoneToUse, sizeof( zoneToUse ), "%s", zone.c_str() );
    status = validateAndParseUserName( userName, userName2, userZone );
    if ( status ) {
        return status;
    }
    if ( userZone[0] != '\0' ) {
        rstrcpy( zoneToUse, userZone, NAME_LEN );
    }
 
    userId[0] = '\0';
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "icatGetTicketUserId SQL 1 " );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( userName2 );
        bindVars.push_back( zoneToUse );
        status = cmlGetStringValueFromSql(
                     "select user_id from R_USER_MAIN where user_name=? and R_USER_MAIN.zone_name=? and user_type_name!='rodsgroup'",
                     userId, NAME_LEN, bindVars, &icss );
    }
    if ( status != 0 ) {
        if ( status == CAT_NO_ROWS_FOUND ) {
            return CAT_INVALID_USER;
        }
        return status;
    }
    strncpy( userIdStr, userId, NAME_LEN );
    return 0;
}
 
int
icatGetTicketGroupId( irods::plugin_property_map& _prop_map, const char *groupName, char *groupIdStr ) {
    char groupId[NAME_LEN];
    char groupZone[NAME_LEN];
    char zoneToUse[NAME_LEN];
    char groupName2[NAME_LEN];
    int status;
 
    std::string zone;
    irods::error ret = getLocalZone( _prop_map, &icss, zone );
    if ( !ret.ok() ) {
        return ret.code();
    }
 
    snprintf( zoneToUse, sizeof( zoneToUse ), "%s", zone.c_str() );
    status = validateAndParseUserName( groupName, groupName2, groupZone );
    if ( status ) {
        return status;
    }
    if ( groupZone[0] != '\0' ) {
        rstrcpy( zoneToUse, groupZone, NAME_LEN );
    }
 
    groupId[0] = '\0';
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "icatGetTicketGroupId SQL 1 " );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( groupName2 );
        bindVars.push_back( zoneToUse );
        status = cmlGetStringValueFromSql(
                     "select user_id from R_USER_MAIN where user_name=? and R_USER_MAIN.zone_name=? and user_type_name='rodsgroup'",
                     groupId, NAME_LEN, bindVars, &icss );
    }
    if ( status != 0 ) {
        if ( status == CAT_NO_ROWS_FOUND ) {
            return CAT_INVALID_GROUP;
        }
        return status;
    }
    strncpy( groupIdStr, groupId, NAME_LEN );
    return 0;
}
 
 
static
int convert_hostname_to_dotted_decimal_ipv4_and_store_in_buffer(const char* _hostname, char* _buf) {
    struct addrinfo hint;
    memset(&hint, 0, sizeof(hint));
    hint.ai_family = AF_INET;
    struct addrinfo *p_addrinfo;
    const int ret_getaddrinfo_with_retry = getaddrinfo_with_retry(_hostname, 0, &hint, &p_addrinfo);
    if (ret_getaddrinfo_with_retry) {
        return ret_getaddrinfo_with_retry;
    }
    sprintf(_buf, "%s", inet_ntoa(reinterpret_cast<struct sockaddr_in*>(p_addrinfo->ai_addr)->sin_addr));
    freeaddrinfo(p_addrinfo);
    return 0;
}
 
 
char *
convertHostToIp( const char *inputName ) {
    static char ipAddr[50];
    const int status = convert_hostname_to_dotted_decimal_ipv4_and_store_in_buffer(inputName, ipAddr);
    if (status != 0) {
        rodsLog( LOG_ERROR, "convertHostToIp convert_hostname_to_dotted_decimal_ipv4_and_store_in_buffer error. status [%d]", status );
        return NULL;
    }
    return ipAddr;
}
 
// XXXX HELPER FUNCTIONS ABOVE
 
// =-=-=-=-=-=-=-
// read a message body off of the socket
irods::error db_start_op(
    irods::plugin_context& _ctx ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    return ret;
 
 
} // db_start_op
 
// =-=-=-=-=-=-=-
// set debug behavior for plugin
irods::error db_debug_op(
    irods::plugin_context& _ctx,
    const char*            _mode ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // check incoming param
    if ( !_mode ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "mode is null" );
    }
 
    // =-=-=-=-=-=-=-
    // run tolower on mode
    std::string mode( _mode );
    std::transform(
        mode.begin(),
        mode.end(),
        mode.begin(),
        ::tolower );
 
    // =-=-=-=-=-=-=-
    // if mode contains 'sql' then turn SQL logging on
    if ( mode.find( "sql" ) != std::string::npos ) {
        logSQL = 1;
    }
    else {
        logSQL = 0;
    }
 
    return SUCCESS();
 
} // db_debug_op
 
// =-=-=-=-=-=-=-
// open a database connection
irods::error db_open_op(
    irods::plugin_context& _ctx ) {
 
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // check incoming param
//        if ( !_cfg ) {
//            return ERROR(
//                       CAT_INVALID_ARGUMENT,
//                       "config is null" );
//        }
 
    // =-=-=-=-=-=-=-
    // log as appropriate
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlOpen" );
    }
 
    // =-=-=-=-=-=-=-
    // cache db creds
    try {
        const auto& db_plugin_map = irods::get_server_property<const std::unordered_map<std::string, boost::any>>(std::vector<std::string>{irods::CFG_PLUGIN_CONFIGURATION_KW, irods::PLUGIN_TYPE_DATABASE});
        const auto& db_type = db_plugin_map.begin()->first;
        const auto& db_plugin = db_plugin_map.begin()->second;
        snprintf(icss.databaseUsername, DB_USERNAME_LEN, "%s", boost::any_cast<const std::string&>(boost::any_cast<const std::unordered_map<std::string, boost::any>>(db_plugin).at(irods::CFG_DB_USERNAME_KW)).c_str());
        snprintf(icss.databasePassword, DB_PASSWORD_LEN, "%s", boost::any_cast<const std::string&>(boost::any_cast<const std::unordered_map<std::string, boost::any>>(db_plugin).at(irods::CFG_DB_PASSWORD_KW)).c_str());
        snprintf(icss.database_plugin_type, DB_TYPENAME_LEN, "%s", db_type.c_str());
    } catch ( const irods::exception& e ) {
        return irods::error(e);
    } catch ( const boost::exception& e ) {
        return ERROR(INVALID_ANY_CAST, "Failed any_cast in the database configuration");
    }
 
    // =-=-=-=-=-=-=-
    // call open in mid level
    int status = cmlOpen( &icss );
    if ( 0 != status ) {
        return ERROR(
                   status,
                   "failed to open db connection" );
    }
 
    // =-=-=-=-=-=-=-
    // set success flag
    icss.status = 1;
 
    // =-=-=-=-=-=-=-
    // Capture ICAT properties
#if MY_ICAT
#elif ORA_ICAT
#else
    irods::catalog_properties::instance().capture_if_needed( &icss );
#endif
 
    // =-=-=-=-=-=-=-
    // set pam properties
    try {
        irods_pam_auth_no_extend = irods::get_server_property<const bool>(std::vector<std::string>{irods::PLUGIN_TYPE_AUTHENTICATION, irods::AUTH_PAM_SCHEME, irods::CFG_PAM_NO_EXTEND_KW});
        irods_pam_password_len = irods::get_server_property<const size_t>(std::vector<std::string>{irods::PLUGIN_TYPE_AUTHENTICATION, irods::AUTH_PAM_SCHEME, irods::CFG_PAM_PASSWORD_LENGTH_KW});
        snprintf(irods_pam_password_min_time, NAME_LEN, "%s", irods::get_server_property<const std::string>(std::vector<std::string>{irods::PLUGIN_TYPE_AUTHENTICATION, irods::AUTH_PAM_SCHEME, irods::CFG_PAM_PASSWORD_MIN_TIME_KW}).c_str());
        snprintf(irods_pam_password_max_time, NAME_LEN, "%s", irods::get_server_property<const std::string>(std::vector<std::string>{irods::PLUGIN_TYPE_AUTHENTICATION, irods::AUTH_PAM_SCHEME, irods::CFG_PAM_PASSWORD_MAX_TIME_KW}).c_str());
    } catch ( const irods::exception& e ) {
        rodsLog(LOG_DEBUG, "[%s:%d] PAM property not found", __FUNCTION__, __LINE__);
        return CODE( status );
    }
 
    if ( irods_pam_auth_no_extend ) {
        snprintf( irods_pam_password_default_time,
                  sizeof( irods_pam_password_default_time ),
                  "%s", "28800" );
    }
 
    return CODE( status );
 
} // db_open_op
 
// =-=-=-=-=-=-=-
// close a database connection
irods::error db_close_op(
    irods::plugin_context& _ctx ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
 
    // =-=-=-=-=-=-=-
    // call open in mid level
    int status = cmlClose( &icss );
    if ( 0 != status ) {
        return ERROR(
                   status,
                   "failed to close db connection" );
    }
 
    // =-=-=-=-=-=-=-
    // set success flag
    icss.status = 0;
 
    return CODE( status );
 
} // db_close_op
 
// =-=-=-=-=-=-=-
// return the local zone
irods::error db_check_and_get_object_id_op(
    irods::plugin_context& _ctx,
    const char*            _type,
    const char*            _name,
    const char*            _access ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    rodsLong_t status = checkAndGetObjectId(
                            _ctx.comm(),
                            _ctx.prop_map(),
                            _type,
                            _name,
                            _access );
    if ( status < 0 ) {
        return ERROR( status, "checkAndGetObjectId failed" );
    }
    else {
        return SUCCESS();
 
    }
 
} // db_check_and_get_object_id_op
 
// =-=-=-=-=-=-=-
// return the local zone
irods::error db_get_local_zone_op(
    irods::plugin_context& _ctx,
    std::string*           _zone ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    ret = getLocalZone( _ctx.prop_map(), &icss, ( *_zone ) );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }
    else {
        return SUCCESS();
 
    }
 
} // db_get_local_zone_op
 
// =-=-=-=-=-=-=-
// update the data obj count of a resource
irods::error db_update_resc_obj_count_op(
    irods::plugin_context& _ctx,
    const std::string*     _resc,
    int                    _delta ) {
 
    return SUCCESS();
 
} // db_update_resc_obj_count_op
 
// =-=-=-=-=-=-=-
// update the data obj count of a resource
irods::error db_mod_data_obj_meta_op(
    irods::plugin_context& _ctx,
    dataObjInfo_t*         _data_obj_info,
    keyValPair_t*          _reg_param ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if ( !_data_obj_info ||
            !_reg_param ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
 
 
    int status = 0, upCols = 0;
    rodsLong_t iVal = 0; // JMC cppcheck - uninit var
    int status2 = 0;
 
    int mode = 0;
 
    char logicalFileName[MAX_NAME_LEN];
    char logicalDirName[MAX_NAME_LEN];
    char *theVal = 0;
    char replNum1[MAX_NAME_LEN];
 
    const char* whereColsAndConds[10];
    const char* whereValues[10];
    char idVal[MAX_NAME_LEN];
    int numConditions = 0;
    char oldCopy[NAME_LEN];
    char newCopy[NAME_LEN];
    std::vector<const char *> updateCols;
    std::vector<const char *> updateVals;
 
    const std::vector<std::string_view> regParamNames = {
        COLL_ID_KW,
        DATA_CREATE_KW,
        CHKSUM_KW,
        DATA_EXPIRY_KW,
        //DATA_ID_KW,
        REPL_STATUS_KW,
        //DATA_MAP_ID_KW,
        DATA_MODE_KW,
        DATA_NAME_KW,
        DATA_OWNER_KW,
        DATA_OWNER_ZONE_KW,
        FILE_PATH_KW,
        REPL_NUM_KW,
        DATA_SIZE_KW,
        STATUS_STRING_KW,
        DATA_TYPE_KW,
        VERSION_KW,
        DATA_MODIFY_KW,
        DATA_COMMENTS_KW,
        // DATA_RESC_GROUP_NAME_KW,
        RESC_HIER_STR_KW,
        RESC_ID_KW,
        RESC_NAME_KW
    };
 
    const std::vector<std::string_view> colNames = {
        "coll_id",
        "create_ts",
        "data_checksum",
        "data_expiry_ts",
        //"data_id",
        "data_is_dirty",
        //"data_map_id",
        "data_mode",
        "data_name",
        "data_owner_name",
        "data_owner_zone",
        "data_path",
        "data_repl_num",
        "data_size",
        "data_status",
        "data_type_name",
        "data_version",
        "modify_ts",
        "r_comment",
        //"resc_group_name",
        "resc_hier",
        "resc_id",
        "resc_name"
    };
 
    int doingDataSize = 0;
    char dataSizeString[NAME_LEN] = "";
    char objIdString[MAX_NAME_LEN];
    char *neededAccess = 0;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlModDataObjMeta" );
    }
 
    bool adminMode{};
    if (getValByKey(_reg_param, ADMIN_KW)) {
        if (LOCAL_PRIV_USER_AUTH != _ctx.comm()->clientUser.authInfo.authFlag) {
            return ERROR(CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "failed with insufficient privilege");
        }
        adminMode = true;
    }
 
    std::string update_resc_id_str;
 
    bool update_resc_id = false;
    /* Set up the updateCols and updateVals arrays */
    std::size_t i = 0, j = 0;
    for (i = 0, j = 0; i < regParamNames.size(); i++) {
        theVal = getValByKey(_reg_param, regParamNames[i].data());
        if (theVal) {
            if(colNames[i] == "resc_name") {
                continue;
            }
            else if(colNames[i] == "resc_hier") {
                updateCols.push_back( "resc_id" );
 
                rodsLong_t resc_id;
                resc_mgr.hier_to_leaf_id(theVal,resc_id);
 
                update_resc_id_str = boost::lexical_cast<std::string>(resc_id);
                updateVals.push_back( update_resc_id_str.c_str() );
            }
            else {
                updateCols.push_back(colNames[i].data());
                updateVals.push_back(theVal);
            }
 
            if ( std::string( "resc_id" ) == colNames[i] || std::string( "resc_hier") == colNames[i]) {
                update_resc_id = true;
            }
 
            if(regParamNames[i] == DATA_EXPIRY_KW) {
                /* if data_expiry, make sure it's in the standard time-stamp format: "%011d" */
                if (colNames[i] == "data_expiry_ts") { /* double check*/
                    if ( strlen( theVal ) < 11 ) {
                        static char theVal2[20];
                        time_t myTimeValue;
                        myTimeValue = atoll( theVal );
                        snprintf( theVal2, sizeof theVal2, "%011d", ( int )myTimeValue );
                        updateVals[j] = theVal2;
                    }
                }
            }
 
            if(regParamNames[i] == DATA_MODIFY_KW) {
                /* if modify_ts, also make sure it's in the standard time-stamp format: "%011d" */
                if (colNames[i] == "modify_ts") { /* double check*/
                    if ( strlen( theVal ) < 11 ) {
                        static char theVal3[20];
                        time_t myTimeValue;
                        myTimeValue = atoll( theVal );
                        snprintf( theVal3, sizeof theVal3, "%011d", ( int )myTimeValue );
                        updateVals[j] = theVal3;
                    }
                }
            }
            if(regParamNames[i] == DATA_SIZE_KW) {
                doingDataSize = 1; /* flag to check size */
                snprintf( dataSizeString, sizeof( dataSizeString ), "%s", theVal );
            }
 
            j++;
 
            /* If the datatype is being updated, check that it is valid */
            if(regParamNames[i] == DATA_TYPE_KW) {
                status = cmlCheckNameToken( "data_type",
                                            theVal, &icss );
                if ( status != 0 ) {
                    std::stringstream msg;
                    msg << __FUNCTION__;
                    msg << " - Invalid data type specified.";
                    addRErrorMsg( &_ctx.comm()->rError, 0, msg.str().c_str() );
                    return ERROR(
                               CAT_INVALID_DATA_TYPE,
                               msg.str() );
                }
            }
        }
    }
    upCols = j;
 
    /* If the only field is the chksum then the user only needs read
       access since we can trust that the server-side code is
       calculating it properly and checksum is a system-managed field.
       For example, when doing an irsync the server may calculate a
       checksum and want to set it in the source copy.
    */
    neededAccess = ACCESS_MODIFY_METADATA;
    if ( upCols == 1 && strcmp( updateCols[0], "chksum" ) == 0 ) {
        neededAccess = ACCESS_READ_OBJECT;
    }
 
    /* If dataExpiry is being updated, user needs to have
       a greater access permission */
    theVal = getValByKey( _reg_param, DATA_EXPIRY_KW );
    if ( theVal != NULL ) {
        neededAccess = ACCESS_DELETE_OBJECT;
    }
 
    if ( _data_obj_info->dataId <= 0 ) {
 
        status = splitPathByKey( _data_obj_info->objPath,
                                 logicalDirName, MAX_NAME_LEN, logicalFileName, MAX_NAME_LEN, '/' );
 
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlModDataObjMeta SQL 1 " );
        }
        {
            std::vector<std::string> bindVars;
            bindVars.push_back( logicalDirName );
            status = cmlGetIntegerValueFromSql(
                         "select coll_id from R_COLL_MAIN where coll_name=?", &iVal,
                         bindVars, &icss );
        }
 
        if ( status != 0 ) {
            char errMsg[105];
            snprintf( errMsg, 100, "collection '%s' is unknown",
                      logicalDirName );
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
            _rollback( "chlModDataObjMeta" );
            return ERROR(
                       CAT_UNKNOWN_COLLECTION,
                       "failed with unknown collection" );
        }
        snprintf( objIdString, MAX_NAME_LEN, "%lld", iVal );
 
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlModDataObjMeta SQL 2" );
        }
        {
            std::vector<std::string> bindVars;
            bindVars.push_back( objIdString );
            bindVars.push_back( logicalFileName );
            status = cmlGetIntegerValueFromSql(
                         "select data_id from R_DATA_MAIN where coll_id=? and data_name=?",
                         &iVal, bindVars, &icss );
        }
        if ( status != 0 ) {
            std::stringstream msg;
            msg << __FUNCTION__;
            msg << " - Failed to find file in database by its logical path.";
            addRErrorMsg( &_ctx.comm()->rError, 0, msg.str().c_str() );
            _rollback( "chlModDataObjMeta" );
            return ERROR(
                       CAT_UNKNOWN_FILE,
                       "failed with unknown file" );
        }
 
        _data_obj_info->dataId = iVal;  /* return it for possible use next time, */
        /* and for use below */
    }
 
    snprintf( objIdString, MAX_NAME_LEN, "%lld", _data_obj_info->dataId );
 
    if (!adminMode) {
        if ( doingDataSize == 1 && strlen( mySessionTicket ) > 0 ) {
            status = cmlTicketUpdateWriteBytes( mySessionTicket,
                                                dataSizeString,
                                                objIdString, &icss );
            if ( status != 0 ) {
                return ERROR(
                           status,
                           "cmlTicketUpdateWriteBytes failed" );
            }
        }
 
        status = cmlCheckDataObjId(
                     objIdString,
                     _ctx.comm()->clientUser.userName,
                     _ctx.comm()->clientUser.rodsZone,
                     neededAccess,
                     mySessionTicket,
                     mySessionClientAddr,
                     &icss );
 
        if ( status != 0 ) {
            theVal = getValByKey( _reg_param, ACL_COLLECTION_KW );
            if ( theVal != NULL && upCols == 1 &&
                    strcmp( updateCols[0], "data_path" ) == 0 ) {
                int len, iVal = 0; // JMC cppcheck - uninit var ( shadows prev decl? )
                /*
                 In this case, the user is doing a 'imv' of a collection but one of
                 the sub-files is not owned by them.  We decided this should be
                 allowed and so we support it via this new ACL_COLLECTION_KW, checking
                 that the ACL_COLLECTION matches the beginning path of the object and
                 that the user has the appropriate access to that collection.
                 */
                len = strlen( theVal );
                if ( strncmp( theVal, _data_obj_info->objPath, len ) == 0 ) {
 
                    iVal = cmlCheckDir( theVal,
                                        _ctx.comm()->clientUser.userName,
                                        _ctx.comm()->clientUser.rodsZone,
                                        ACCESS_OWN,
                                        &icss );
                }
                if ( iVal > 0 ) {
                    status = 0;
                } /* Collection was found (id
                                   * returned) & user has access */
            }
            if ( status ) {
                _rollback( "chlModDataObjMeta" );
                return ERROR(
                           CAT_NO_ACCESS_PERMISSION,
                           "failed with no permission" );
            }
        }
    }
 
    whereColsAndConds[0] = "data_id=";
    snprintf( idVal, MAX_NAME_LEN, "%lld", _data_obj_info->dataId );
    whereValues[0] = idVal;
    numConditions = 1;
 
    /* This is up here since this is usually called to modify the
     * metadata of a single repl.  If ALL_KW is included, then apply
     * this change to all replicas (by not restricting the update to
     * only one).
     */
    std::string where_resc_id_str;
    if ( getValByKey( _reg_param, ALL_KW ) == NULL ) {
        // use resc_id instead of replNum as it is
        // always set, unless resc_id is to be
        // updated.  replNum is sometimes 0 in various
        // error cases
        if ( update_resc_id || strlen( _data_obj_info->rescHier ) <= 0 ) {
            j = numConditions;
            whereColsAndConds[j] = "data_repl_num=";
            snprintf( replNum1, MAX_NAME_LEN, "%d", _data_obj_info->replNum );
            whereValues[j] = replNum1;
            numConditions++;
 
        }
        else {
            rodsLong_t id = 0;
            resc_mgr.hier_to_leaf_id( _data_obj_info->rescHier, id );
            where_resc_id_str = boost::lexical_cast<std::string>(id);
            j = numConditions;
            whereColsAndConds[j] = "resc_id=";
            whereValues[j] = where_resc_id_str.c_str();
            numConditions++;
        }
    }
 
    mode = 0;
    if ( getValByKey( _reg_param, ALL_REPL_STATUS_KW ) ) {
        mode = 1;
        /* mark this one as NEWLY_CREATED_COPY and others as OLD_COPY */
    }
 
    std::string zone;
    ret = getLocalZone(
              _ctx.prop_map(),
              &icss,
              zone );
    if ( !ret.ok() ) {
        rodsLog( LOG_ERROR, "chlModObjMeta - failed in getLocalZone with status [%d]", status );
        return PASS( ret );
    }
 
    if ( mode == 0 ) {
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlModDataObjMeta SQL 4" );
        }
        status = cmlModifySingleTable(
                     "R_DATA_MAIN",
                     &( updateCols[0] ),
                     &( updateVals[0] ),
                     whereColsAndConds,
                     whereValues,
                     upCols,
                     numConditions,
                     &icss );
    }
    else {
        /* mark this one as NEWLY_CREATED_COPY and others as OLD_COPY */
        j = upCols;
        updateCols.push_back( "data_is_dirty" );
        snprintf( newCopy, NAME_LEN, "%d", NEWLY_CREATED_COPY );
        updateVals.push_back( newCopy );
        upCols++;
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlModDataObjMeta SQL 5" );
        }
        status = cmlModifySingleTable(
                     "R_DATA_MAIN",
                     &( updateCols[0] ),
                     &( updateVals[0] ),
                     whereColsAndConds,
                     whereValues,
                     upCols,
                     numConditions,
                     &icss );
        if ( status == 0 ) {
            j = numConditions - 1;
            whereColsAndConds[j] = "data_repl_num!=";
            snprintf( replNum1, MAX_NAME_LEN, "%d", _data_obj_info->replNum );
            whereValues[j] = replNum1;
 
            updateCols[0] = "data_is_dirty";
            snprintf( oldCopy, NAME_LEN, "%d", OLD_COPY );
            updateVals[0] = oldCopy;
            if ( logSQL != 0 ) {
                rodsLog( LOG_SQL, "chlModDataObjMeta SQL 6" );
            }
            status2 = cmlModifySingleTable( "R_DATA_MAIN", &( updateCols[0] ), &( updateVals[0] ),
                                            whereColsAndConds, whereValues, 1,
                                            numConditions, &icss );
 
            if ( status2 != 0 && status2 != CAT_SUCCESS_BUT_WITH_NO_INFO ) {
                /* Ignore NO_INFO errors but not others */
                rodsLog( LOG_NOTICE,
                         "chlModDataObjMeta cmlModifySingleTable failure for other replicas %d",
                         status2 );
                _rollback( "chlModDataObjMeta" );
                return ERROR(
                           status2,
                           "cmlModifySingleTable failure for other replicas" );
            }
 
        }
    }
    if ( status != 0 ) {
        _rollback( "chlModDataObjMeta" );
        rodsLog( LOG_NOTICE,
                 "chlModDataObjMeta cmlModifySingleTable failure %d",
                 status );
        return ERROR(
                   status,
                   "cmlModifySingleTable failure" );
    }
 
    if ( !( _data_obj_info->flags & NO_COMMIT_FLAG ) ) {
        status =  cmlExecuteNoAnswerSql( "commit", &icss );
        if ( status != 0 ) {
            rodsLog( LOG_NOTICE,
                     "chlModDataObjMeta cmlExecuteNoAnswerSql commit failure %d",
                     status );
            return ERROR(
                       status,
                       "commit failure" );
        }
    }
 
    return CODE( status );
 
} // db_mod_data_obj_meta_op
 
 
// =-=-=-=-=-=-=-
// update the data obj count of a resource
irods::error db_reg_data_obj_op(
    irods::plugin_context& _ctx,
    dataObjInfo_t*         _data_obj_info ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if ( !_data_obj_info ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
 
    // =-=-=-=-=-=-=-
    //
    char myTime[50];
    char logicalFileName[MAX_NAME_LEN];
    char logicalDirName[MAX_NAME_LEN];
    rodsLong_t seqNum;
    rodsLong_t iVal;
    char dataIdNum[MAX_NAME_LEN];
    char collIdNum[MAX_NAME_LEN];
    char dataReplNum[MAX_NAME_LEN];
    char dataSizeNum[MAX_NAME_LEN];
    char dataStatusNum[MAX_NAME_LEN];
    char data_expiry_ts[] = { "00000000000" };
    int status;
    int inheritFlag;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegDataObj" );
    }
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegDataObj SQL 1 " );
    }
    seqNum = cmlGetNextSeqVal( &icss );
    if ( seqNum < 0 ) {
        rodsLog( LOG_NOTICE, "chlRegDataObj cmlGetNextSeqVal failure %d",
                 seqNum );
        _rollback( "chlRegDataObj" );
        return ERROR( seqNum, "chlRegDataObj cmlGetNextSeqVal failure" );
    }
    snprintf( dataIdNum, MAX_NAME_LEN, "%lld", seqNum );
    _data_obj_info->dataId = seqNum; /* store as output parameter */
 
    status = splitPathByKey( _data_obj_info->objPath,
                             logicalDirName, MAX_NAME_LEN, logicalFileName, MAX_NAME_LEN, '/' );
 
 
    /* Check that collection exists and user has write permission.
       At the same time, also get the inherit flag */
    iVal = cmlCheckDirAndGetInheritFlag( logicalDirName,
                                         _ctx.comm()->clientUser.userName,
                                         _ctx.comm()->clientUser.rodsZone,
                                         ACCESS_MODIFY_OBJECT,
                                         &inheritFlag,
                                         mySessionTicket,
                                         mySessionClientAddr,
                                         &icss );
    if ( iVal < 0 ) {
        if ( iVal == CAT_UNKNOWN_COLLECTION ) {
            std::stringstream errMsg;
            errMsg << "collection '" << logicalDirName << "' is unknown";
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg.str().c_str() );
        }
        else if ( iVal == CAT_NO_ACCESS_PERMISSION ) {
            std::stringstream errMsg;
            errMsg << "no permission to update collection '" << logicalDirName << "'";
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg.str().c_str() );
        }
        return ERROR( iVal, "" );
    }
    snprintf( collIdNum, MAX_NAME_LEN, "%lld", iVal );
 
    /* Make sure no collection already exists by this name */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegDataObj SQL 4" );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( _data_obj_info->objPath );
        status = cmlGetIntegerValueFromSql(
                     "select coll_id from R_COLL_MAIN where coll_name=?",
                     &iVal, bindVars, &icss );
    }
    if ( status == 0 ) {
        return ERROR( CAT_NAME_EXISTS_AS_COLLECTION, "collection exists" );
    }
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegDataObj SQL 5" );
    }
    status = cmlCheckNameToken( "data_type",
                                _data_obj_info->dataType, &icss );
    if ( status != 0 ) {
        return ERROR( CAT_INVALID_DATA_TYPE, "invalid data type" );
    }
 
    snprintf( dataReplNum, MAX_NAME_LEN, "%d", _data_obj_info->replNum );
    snprintf( dataStatusNum, MAX_NAME_LEN, "%d", _data_obj_info->replStatus );
    snprintf( dataSizeNum, MAX_NAME_LEN, "%lld", _data_obj_info->dataSize );
    getNowStr( myTime );
 
    if (0 == strcmp(_data_obj_info->dataModify, "")) {
        strcpy(_data_obj_info->dataModify, myTime);
    }
 
    std::string resc_id_str = boost::lexical_cast<std::string>(_data_obj_info->rescId);
    cllBindVars[0] = dataIdNum;
    cllBindVars[1] = collIdNum;
    cllBindVars[2] = logicalFileName;
    cllBindVars[3] = dataReplNum;
    cllBindVars[4] = _data_obj_info->version;
    cllBindVars[5] = _data_obj_info->dataType;
    cllBindVars[6] = dataSizeNum;
    cllBindVars[7] = resc_id_str.c_str();
    cllBindVars[8] = _data_obj_info->filePath;
    cllBindVars[9] = _ctx.comm()->clientUser.userName;
    cllBindVars[10] = _ctx.comm()->clientUser.rodsZone;
    cllBindVars[11] = dataStatusNum;
    cllBindVars[12] = _data_obj_info->chksum;
    cllBindVars[13] = _data_obj_info->dataMode;
    cllBindVars[14] = myTime;
    cllBindVars[15] = _data_obj_info->dataModify;
    cllBindVars[16] = data_expiry_ts;
    cllBindVars[17] = "EMPTY_RESC_NAME";
    cllBindVars[18] = "EMPTY_RESC_HIER";
    cllBindVars[19] = "EMPTY_RESC_GROUP_NAME";
    cllBindVarCount = 20;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegDataObj SQL 6" );
    }
    status =  cmlExecuteNoAnswerSql(
                  "insert into R_DATA_MAIN (data_id, coll_id, data_name, data_repl_num, data_version, data_type_name, data_size, resc_id, data_path, data_owner_name, data_owner_zone, data_is_dirty, data_checksum, data_mode, create_ts, modify_ts, data_expiry_ts, resc_name, resc_hier, resc_group_name) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                  &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegDataObj cmlExecuteNoAnswerSql failure %d", status );
        _rollback( "chlRegDataObj" );
        return ERROR( status, "chlRegDataObj cmlExecuteNoAnswerSql failure" );
    }
    std::string zone;
    ret = getLocalZone(
              _ctx.prop_map(),
              &icss,
              zone );
    if ( !ret.ok() ) {
        rodsLog( LOG_ERROR, "chlRegDataInfo - failed in getLocalZone with status [%d]", status );
        return PASS( ret );
    }
 
    if ( inheritFlag ) {
        /* If inherit is set (sticky bit), then add access rows for this
           dataobject that match those of the parent collection */
        cllBindVars[0] = dataIdNum;
        cllBindVars[1] = myTime;
        cllBindVars[2] = myTime;
        cllBindVars[3] = collIdNum;
        cllBindVarCount = 4;
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlRegDataObj SQL 7" );
        }
        status =  cmlExecuteNoAnswerSql(
                      "insert into R_OBJT_ACCESS (object_id, user_id, access_type_id, create_ts, modify_ts) (select ?, user_id, access_type_id, ?, ? from R_OBJT_ACCESS where object_id = ?)",
                      &icss );
        if ( status != 0 ) {
            rodsLog( LOG_NOTICE,
                     "chlRegDataObj cmlExecuteNoAnswerSql insert access failure %d",
                     status );
            _rollback( "chlRegDataObj" );
            return ERROR( status, "cmlExecuteNoAnswerSql insert access failure" );
        }
    }
    else {
        cllBindVars[0] = dataIdNum;
        cllBindVars[1] = _ctx.comm()->clientUser.userName;
        cllBindVars[2] = _ctx.comm()->clientUser.rodsZone;
        cllBindVars[3] = ACCESS_OWN;
        cllBindVars[4] = myTime;
        cllBindVars[5] = myTime;
        cllBindVarCount = 6;
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlRegDataObj SQL 8" );
        }
        status =  cmlExecuteNoAnswerSql(
                      "insert into R_OBJT_ACCESS values (?, (select user_id from R_USER_MAIN where user_name=? and zone_name=?), (select token_id from R_TOKN_MAIN where token_namespace = 'access_type' and token_name = ?), ?, ?)",
                      &icss );
        if ( status != 0 ) {
            rodsLog( LOG_NOTICE,
                     "chlRegDataObj cmlExecuteNoAnswerSql insert access failure %d",
                     status );
            _rollback( "chlRegDataObj" );
            return ERROR( status, "cmlExecuteNoAnswerSql insert access failure" );
        }
    }
 
    status = cmlAudit3( AU_REGISTER_DATA_OBJ, dataIdNum,
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone, "", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegDataObj cmlAudit3 failure %d",
                 status );
        _rollback( "chlRegDataObj" );
        return ERROR( status, "cmlAudit3 failure" );
    }
 
 
    if ( !( _data_obj_info->flags & NO_COMMIT_FLAG ) ) {
        status =  cmlExecuteNoAnswerSql( "commit", &icss );
        if ( status != 0 ) {
            rodsLog( LOG_NOTICE,
                     "chlRegDataObj cmlExecuteNoAnswerSql commit failure %d",
                     status );
            return ERROR( status, "cmlExecuteNoAnswerSql commit failure" );
        }
    }
 
    return SUCCESS();
 
} // db_reg_data_obj_op
 
 
// =-=-=-=-=-=-=-
// register a data object into the catalog
irods::error db_reg_replica_op(
    irods::plugin_context& _ctx,
    dataObjInfo_t*         _src_data_obj_info,
    dataObjInfo_t*         _dst_data_obj_info,
    keyValPair_t*          _cond_input ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_src_data_obj_info ||
        !_dst_data_obj_info ||
        !_cond_input ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    if( _dst_data_obj_info->rescId <= 0 ) {
        std::stringstream msg;
        msg << "invalid resource id "
            << _dst_data_obj_info->rescId
            << " for ["
            << _dst_data_obj_info->objPath
            << "]";
        return ERROR(
                SYS_INVALID_INPUT_PARAM,
                msg.str() );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
 
 
    char myTime[50];
    char logicalFileName[MAX_NAME_LEN];
    char logicalDirName[MAX_NAME_LEN];
    rodsLong_t iVal;
    rodsLong_t status;
    char tSQL[MAX_SQL_SIZE];
    char *cVal[30];
    int i;
    int statementNumber = UNINITIALIZED_STATEMENT_NUMBER;
    int nextReplNum;
    char nextRepl[30];
    char theColls[] = "data_id, \
                       coll_id,  \
                       data_name, \
                       data_repl_num, \
                       data_version, \
                       data_type_name, \
                       data_size, \
                       resc_group_name, \
                       resc_name, \
                       resc_hier, \
                       resc_id, \
                       data_path, \
                       data_owner_name, \
                       data_owner_zone, \
                       data_is_dirty, \
                       data_status, \
                       data_checksum, \
                       data_expiry_ts, \
                       data_map_id, \
                       data_mode, \
                       r_comment, \
                       create_ts, \
                       modify_ts";
    const int IX_DATA_REPL_NUM = 3; /* index of data_repl_num in theColls */
//        int IX_RESC_GROUP_NAME = 7; /* index into theColls */
    const int IX_RESC_ID = 10;
    const int IX_DATA_PATH = 11;    /* index into theColls */
 
    const int IX_DATA_MODE = 19;
    const int IX_CREATE_TS = 21;
    const int IX_MODIFY_TS = 22;
    const int IX_RESC_NAME2 = 23;
    const int IX_DATA_PATH2 = 24;
    const int IX_DATA_ID2 = 25;
    int nColumns = 26;
 
    char objIdString[MAX_NAME_LEN];
    char replNumString[MAX_NAME_LEN];
    int adminMode;
    char *theVal;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegReplica" );
    }
 
    adminMode = 0;
    if ( _cond_input != NULL ) {
        theVal = getValByKey( _cond_input, ADMIN_KW );
        if ( theVal != NULL ) {
            adminMode = 1;
        }
    }
 
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    status = splitPathByKey( _src_data_obj_info->objPath,
                             logicalDirName, MAX_NAME_LEN, logicalFileName, MAX_NAME_LEN, '/' );
 
    if ( adminMode ) {
        if ( _ctx.comm()->clientUser.authInfo.authFlag != LOCAL_PRIV_USER_AUTH ) {
            return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege" );
        }
    }
    else {
        /* Check the access to the dataObj */
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlRegReplica SQL 1 " );
        }
        status = cmlCheckDataObjOnly( logicalDirName, logicalFileName,
                                      _ctx.comm()->clientUser.userName,
                                      _ctx.comm()->clientUser.rodsZone,
                                      ACCESS_READ_OBJECT, &icss );
        if ( status < 0 ) {
            _rollback( "chlRegReplica" );
            return ERROR( status, "cmlCheckDataObjOnly failed" );
        }
    }
 
    /* Get the next replica number */
    snprintf( objIdString, MAX_NAME_LEN, "%lld", _src_data_obj_info->dataId );
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegReplica SQL 2" );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( objIdString );
        status = cmlGetIntegerValueFromSql(
                     "select max(data_repl_num) from R_DATA_MAIN where data_id = ?",
                     &iVal, bindVars, &icss );
    }
 
    if ( status != 0 ) {
        _rollback( "chlRegReplica" );
        return ERROR( status, "cmlGetIntegerValueFromSql failed" );
    }
 
    nextReplNum = iVal + 1;
    snprintf( nextRepl, sizeof nextRepl, "%d", nextReplNum );
    _dst_data_obj_info->replNum = nextReplNum; /* return new replica number */
    snprintf( replNumString, MAX_NAME_LEN, "%d", _src_data_obj_info->replNum );
    snprintf( tSQL, MAX_SQL_SIZE,
              "select %s from R_DATA_MAIN where data_id = ? and data_repl_num = ?",
              theColls );
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegReplica SQL 3" );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( objIdString );
        bindVars.push_back( replNumString );
        status = cmlGetOneRowFromSqlV2( tSQL, cVal, nColumns, bindVars, &icss );
    }
    if ( status < 0 ) {
        _rollback( "chlRegReplica" );
        return ERROR( status, "cmlGetOneRowFromSqlV2 failed" );
    }
    statementNumber = status;
 
    std::string resc_id_str = boost::lexical_cast<std::string>(_dst_data_obj_info->rescId);
 
    cVal[IX_DATA_REPL_NUM]   = nextRepl;
    //cVal[IX_RESC_NAME]       = _dst_data_obj_info->rescName;
    cVal[IX_RESC_ID]       = (char*)resc_id_str.c_str();
    cVal[IX_DATA_PATH]       = _dst_data_obj_info->filePath;
    cVal[IX_DATA_MODE]       = _dst_data_obj_info->dataMode;
 
    getNowStr( myTime );
    cVal[IX_MODIFY_TS] = myTime;
    cVal[IX_CREATE_TS] = myTime;
 
    cVal[IX_RESC_NAME2] = (char*)resc_id_str.c_str();//_dst_data_obj_info->rescName; // JMC - backport 4669
    cVal[IX_DATA_PATH2] = _dst_data_obj_info->filePath; // JMC - backport 4669
    cVal[IX_DATA_ID2] = objIdString; // JMC - backport 4669
 
    for ( i = 0; i < nColumns; i++ ) {
        cllBindVars[i] = cVal[i];
    }
    cllBindVarCount = nColumns;
#if (defined ORA_ICAT || defined MY_ICAT) // JMC - backport 4685
    /* MySQL and Oracle */
    snprintf( tSQL, MAX_SQL_SIZE, "insert into R_DATA_MAIN ( %s ) select ?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,? from DUAL where not exists (select data_id from R_DATA_MAIN where resc_id=? and data_path=? and data_id=?)",
              theColls );
#else
    /* Postgres */
    snprintf( tSQL, MAX_SQL_SIZE, "insert into R_DATA_MAIN ( %s ) select ?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,? where not exists (select data_id from R_DATA_MAIN where resc_id=? and data_path=? and data_id=?)",
              theColls );
 
#endif
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegReplica SQL 4" );
    }
    status = cmlExecuteNoAnswerSql( tSQL,  &icss );
    if ( status < 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegReplica cmlExecuteNoAnswerSql(insert) failure %d",
                 status );
        _rollback( "chlRegReplica" );
        const int free_status = cmlFreeStatement( statementNumber, &icss );
        if (free_status != 0) {
            rodsLog(LOG_ERROR, "db_reg_replica_op: cmlFreeStatement0 failure [%d]", free_status);
        }
        return ERROR( status, "cmlExecuteNoAnswerSql(insert) failure" );
    }
 
    std::string zone;
    ret = getLocalZone( _ctx.prop_map(), &icss, zone );
    if ( !ret.ok() ) {
        rodsLog( LOG_ERROR, "chlRegReplica - failed in getLocalZone with status [%d]", status );
        const int free_status = cmlFreeStatement( statementNumber, &icss );
        if (free_status != 0) {
            rodsLog(LOG_ERROR, "db_reg_replica_op: cmlFreeStatement1 failure [%d]", free_status);
        }
        return PASS( ret );
    }
 
    status = cmlFreeStatement( statementNumber, &icss );
    if ( status < 0 ) {
        rodsLog( LOG_NOTICE, "chlRegReplica cmlFreeStatement failure %d", status );
        return ERROR( status, "cmlFreeStatement failure" );
    }
 
    status = cmlAudit3( AU_REGISTER_DATA_REPLICA, objIdString,
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone, nextRepl, &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegDataReplica cmlAudit3 failure %d",
                 status );
        _rollback( "chlRegReplica" );
        return ERROR( status, "cmlAudit3 failure" );
    }
 
    status =  cmlExecuteNoAnswerSql( "commit", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegReplica cmlExecuteNoAnswerSql commit failure %d",
                 status );
        return ERROR( status, "cmlExecuteNoAnswerSql commit failure" );
    }
 
    return SUCCESS();
 
} // db_reg_replica_op
 
// =-=-=-=-=-=-=-
// unregister a data object
irods::error db_unreg_replica_op(
    irods::plugin_context& _ctx,
    dataObjInfo_t*         _data_obj_info,
    keyValPair_t*          _cond_input ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_data_obj_info ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
 
 
    // =-=-=-=-=-=-=-
    //
    char logicalFileName[MAX_NAME_LEN];
    char logicalDirName[MAX_NAME_LEN];
    rodsLong_t status;
    char tSQL[MAX_SQL_SIZE];
    char replNumber[30];
    char dataObjNumber[30];
    int adminMode;
    int trashMode;
    char *theVal;
    char checkPath[MAX_NAME_LEN];
 
    dataObjNumber[0] = '\0';
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlUnregDataObj" );
    }
 
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    adminMode = 0;
    trashMode = 0;
    if ( _cond_input != NULL ) {
        theVal = getValByKey( _cond_input, ADMIN_KW );
        if ( theVal != NULL ) {
            adminMode = 1;
        }
        theVal = getValByKey( _cond_input, ADMIN_RMTRASH_KW );
        if ( theVal != NULL ) {
            adminMode = 1;
            trashMode = 1;
        }
    }
 
    status = splitPathByKey( _data_obj_info->objPath,
                             logicalDirName, MAX_NAME_LEN, logicalFileName, MAX_NAME_LEN, '/' );
 
 
    if ( adminMode == 0 ) {
        /* Check the access to the dataObj */
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlUnregDataObj SQL 1 " );
        }
        status = cmlCheckDataObjOnly( logicalDirName, logicalFileName,
                                      _ctx.comm()->clientUser.userName,
                                      _ctx.comm()->clientUser.rodsZone,
                                      ACCESS_DELETE_OBJECT, &icss );
        if ( status < 0 ) {
            _rollback( "chlUnregDataObj" );
            return ERROR( status, "cmlCheckDataObjOnly failed" ); /* convert long to int */
        }
        snprintf( dataObjNumber, sizeof dataObjNumber, "%lld", status );
    }
    else {
        if ( _ctx.comm()->clientUser.authInfo.authFlag != LOCAL_PRIV_USER_AUTH ) {
            return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege" );
        }
        if ( trashMode ) {
            int len;
            std::string zone;
            ret = getLocalZone( _ctx.prop_map(), &icss, zone );
            if ( !ret.ok() ) {
                return PASS( ret );
            }
            snprintf( checkPath, MAX_NAME_LEN, "/%s/trash", zone.c_str() );
            len = strlen( checkPath );
            if ( strncmp( checkPath, logicalDirName, len ) != 0 ) {
                addRErrorMsg( &_ctx.comm()->rError, 0,
                              "TRASH_KW but not zone/trash path" );
                return ERROR( CAT_INVALID_ARGUMENT, "TRASH_KW but not zone/trash path" );
            }
            if ( _data_obj_info->dataId > 0 ) {
                snprintf( dataObjNumber, sizeof dataObjNumber, "%lld",
                          _data_obj_info->dataId );
            }
        }
        else {
            if ( _data_obj_info->replNum >= 0 && _data_obj_info->dataId >= 0 ) {
                snprintf( dataObjNumber, sizeof dataObjNumber, "%lld",
                          _data_obj_info->dataId );
            }
            else {
                addRErrorMsg( &_ctx.comm()->rError, 0,
                              "dataId and replNum required" );
                _rollback( "chlUnregDataObj" );
                return ERROR( CAT_INVALID_ARGUMENT, "dataId and replNum required" );
            }
        }
    }
 
    // Get the resource name so we can update its data object count later
    std::string resc_hier;
    if ( strlen( _data_obj_info->rescHier ) == 0 ) {
        rodsLong_t resc_id = 0;
        if ( _data_obj_info->replNum >= 0 ) {
            snprintf( replNumber, sizeof replNumber, "%d", _data_obj_info->replNum );
            {
                std::vector<std::string> bindVars;
                bindVars.push_back( dataObjNumber );
                bindVars.push_back( replNumber );
                status = cmlGetIntegerValueFromSql(
                             "select resc_id from R_DATA_MAIN where data_id=? and data_repl_num=?",
                             &resc_id, bindVars, &icss );
            }
            if ( status != 0 ) {
                return ERROR( status, "cmlGetStringValueFromSql failed" );
            }
        }
        else {
            {
                std::vector<std::string> bindVars;
                bindVars.push_back( dataObjNumber );
                status = cmlGetIntegerValueFromSql(
                             "select resc_id from R_DATA_MAIN where data_id=? and data_repl_num=?",
                             &resc_id, bindVars, &icss );
            }
            if ( status != 0 ) {
                return ERROR( status, "cmlGetStringValueFromSql failed" );
            }
        }
        resc_mgr.leaf_id_to_hier( resc_id, resc_hier );
    }
    else {
        resc_hier = std::string( _data_obj_info->rescHier );
    }
 
    cllBindVars[0] = logicalDirName;
    cllBindVars[1] = logicalFileName;
    if ( _data_obj_info->replNum >= 0 ) {
        snprintf( replNumber, sizeof replNumber, "%d", _data_obj_info->replNum );
        cllBindVars[2] = replNumber;
        cllBindVarCount = 3;
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlUnregDataObj SQL 4" );
        }
        snprintf( tSQL, MAX_SQL_SIZE,
                  "delete from R_DATA_MAIN where coll_id=(select coll_id from R_COLL_MAIN where coll_name=?) and data_name=? and data_repl_num=?" );
    }
    else {
        cllBindVarCount = 2;
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlUnregDataObj SQL 5" );
        }
        snprintf( tSQL, MAX_SQL_SIZE,
                  "delete from R_DATA_MAIN where coll_id=(select coll_id from R_COLL_MAIN where coll_name=?) and data_name=?" );
    }
    status =  cmlExecuteNoAnswerSql( tSQL, &icss );
    if ( status != 0 ) {
        if ( status == CAT_SUCCESS_BUT_WITH_NO_INFO ) {
            char errMsg[105];
            status = CAT_UNKNOWN_FILE;  /* More accurate, in this case */
            snprintf( errMsg, 100, "data object '%s' is unknown",
                      logicalFileName );
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
            return ERROR( status, "data object unknown" );
        }
        _rollback( "chlUnregDataObj" );
        return ERROR( status, "cmlExecuteNoAnswerSql failed" );
    }
 
    std::string zone;
    ret = getLocalZone( _ctx.prop_map(), &icss, zone );
    if ( !ret.ok() ) {
        rodsLog( LOG_ERROR, "chlUnRegDataObj - failed in getLocalZone with status [%d]", status );
        return PASS( ret );
    }
 
    /* delete the access rows, if we just deleted the last replica */
    if ( dataObjNumber[0] != '\0' ) {
        cllBindVars[0] = dataObjNumber;
        cllBindVars[1] = dataObjNumber;
        cllBindVarCount = 2;
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlUnregDataObj SQL 3" );
        }
        status = cmlExecuteNoAnswerSql(
                     "delete from R_OBJT_ACCESS where object_id=? and not exists (select * from R_DATA_MAIN where data_id=?)", &icss );
        if ( status == 0 ) {
            removeMetaMapAndAVU( dataObjNumber ); /* remove AVU metadata, if any */
        }
    }
 
    /* Audit */
    if ( dataObjNumber[0] != '\0' ) {
        status = cmlAudit3( AU_UNREGISTER_DATA_OBJ, dataObjNumber,
                            _ctx.comm()->clientUser.userName,
                            _ctx.comm()->clientUser.rodsZone, "", &icss );
    }
    else {
        status = cmlAudit3( AU_UNREGISTER_DATA_OBJ, "0",
                            _ctx.comm()->clientUser.userName,
                            _ctx.comm()->clientUser.rodsZone,
                            _data_obj_info->objPath, &icss );
    }
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlUnregDataObj cmlAudit3 failure %d",
                 status );
        _rollback( "chlUnregDataObj" );
        return ERROR( status, "cmlAudit3 failure" );
    }
 
 
    status =  cmlExecuteNoAnswerSql( "commit", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlUnregDataObj cmlExecuteNoAnswerSql commit failure %d",
                 status );
        return ERROR( status, "cmlExecuteNoAnswerSql commit failure" );
    }
 
    return SUCCESS();
 
} // db_unreg_replica_op
 
// =-=-=-=-=-=-=-
//
irods::error db_reg_rule_exec_op(
    irods::plugin_context& _ctx,
    ruleExecSubmitInp_t*   _re_sub_inp ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_re_sub_inp ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
 
    char myTime[50];
    rodsLong_t seqNum;
    char ruleExecIdNum[MAX_NAME_LEN];
    int status;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegRuleExec" );
    }
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegRuleExec SQL 1 " );
    }
    seqNum = cmlGetNextSeqVal( &icss );
    if ( seqNum < 0 ) {
        rodsLog( LOG_NOTICE, "chlRegRuleExec cmlGetNextSeqVal failure %d",
                 seqNum );
        _rollback( "chlRegRuleExec" );
        return ERROR( seqNum, "cmlGetNextSeqVal failure" );
    }
    snprintf( ruleExecIdNum, MAX_NAME_LEN, "%lld", seqNum );
 
    /* store as output parameter */
    snprintf( _re_sub_inp->ruleExecId, NAME_LEN, "%s", ruleExecIdNum );
 
    getNowStr( myTime );
 
    cllBindVars[0] = ruleExecIdNum;
    cllBindVars[1] = _re_sub_inp->ruleName;
    cllBindVars[2] = _re_sub_inp->reiFilePath;
    cllBindVars[3] = _re_sub_inp->userName;
    cllBindVars[4] = _re_sub_inp->exeAddress;
    cllBindVars[5] = _re_sub_inp->exeTime;
    cllBindVars[6] = _re_sub_inp->exeFrequency;
    cllBindVars[7] = _re_sub_inp->priority;
    cllBindVars[8] = _re_sub_inp->estimateExeTime;
    cllBindVars[9] = _re_sub_inp->notificationAddr;
    cllBindVars[10] = myTime;
    cllBindVars[11] = myTime;
 
    cllBindVarCount = 12;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegRuleExec SQL 2" );
    }
    status =  cmlExecuteNoAnswerSql(
                  "insert into R_RULE_EXEC (rule_exec_id, rule_name, rei_file_path, user_name, exe_address, exe_time, exe_frequency, priority, estimated_exe_time, notification_addr, create_ts, modify_ts) values (?,?,?,?,?,?,?,?,?,?,?,?)",
                  &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegRuleExec cmlExecuteNoAnswerSql(insert) failure %d", status );
        _rollback( "chlRegRuleExec" );
        return ERROR( status, "cmlExecuteNoAnswerSql(insert) failure" );
 
    }
 
    /* Audit */
    status = cmlAudit3( AU_REGISTER_DELAYED_RULE,  ruleExecIdNum,
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone,
                        _re_sub_inp->ruleName, &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegRuleExec cmlAudit3 failure %d",
                 status );
        _rollback( "chlRegRuleExec" );
        return ERROR( status, "cmlAudit3 failure" );
    }
 
    status =  cmlExecuteNoAnswerSql( "commit", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegRuleExec cmlExecuteNoAnswerSql commit failure %d",
                 status );
        return ERROR( status, "cmlExecuteNoAnswerSql commit failure" );
    }
 
    return SUCCESS();
 
} // db_reg_rule_exec_op
 
// =-=-=-=-=-=-=-
// unregister a data object
irods::error db_mod_rule_exec_op(
    irods::plugin_context& _ctx,
    const char*            _re_id,
    keyValPair_t*          _reg_param ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_re_id  ||
        !_reg_param ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
 
 
    // =-=-=-=-=-=-=-
    //
    int i, j, status;
 
    char tSQL[MAX_SQL_SIZE];
    char *theVal = 0;
 
    /* regParamNames has the argument names (in regParam) that this
       routine understands and colNames has the corresponding column
       names; one for one. */
    char *regParamNames[] = {
        RULE_NAME_KW, RULE_REI_FILE_PATH_KW, RULE_USER_NAME_KW,
        RULE_EXE_ADDRESS_KW, RULE_EXE_TIME_KW,
        RULE_EXE_FREQUENCY_KW, RULE_PRIORITY_KW, RULE_ESTIMATE_EXE_TIME_KW,
        RULE_NOTIFICATION_ADDR_KW, RULE_LAST_EXE_TIME_KW,
        RULE_EXE_STATUS_KW,
        "END"
    };
    char *colNames[] = {
        "rule_name", "rei_file_path", "user_name",
        "exe_address", "exe_time", "exe_frequency", "priority",
        "estimated_exe_time", "notification_addr",
        "last_exe_time", "exe_status",
        "create_ts", "modify_ts",
    };
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlModRuleExec" );
    }
 
    snprintf( tSQL, MAX_SQL_SIZE, "update R_RULE_EXEC set " );
 
    for ( i = 0, j = 0; strcmp( regParamNames[i], "END" ); i++ ) {
        theVal = getValByKey( _reg_param, regParamNames[i] );
        if ( theVal != NULL ) {
            if ( j > 0 ) {
                rstrcat( tSQL, "," , MAX_SQL_SIZE );
            }
            rstrcat( tSQL, colNames[i] , MAX_SQL_SIZE );
            rstrcat( tSQL, "=? ", MAX_SQL_SIZE );
            cllBindVars[j++] = theVal;
        }
    }
 
    if ( j == 0 ) {
        return ERROR( CAT_INVALID_ARGUMENT, "invalid argument" );
    }
 
    rstrcat( tSQL, "where rule_exec_id=?", MAX_SQL_SIZE );
    cllBindVars[j++] = _re_id;
    cllBindVarCount = j;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlModRuleExec SQL 1 " );
    }
    status =  cmlExecuteNoAnswerSql( tSQL, &icss );
 
    if ( status != 0 ) {
        _rollback( "chlModRuleExec" );
        rodsLog( LOG_NOTICE,
                 "chlModRuleExec cmlExecuteNoAnswer(update) failure %d",
                 status );
        return ERROR( status, "cmlExecuteNoAnswer(update) failure" );
    }
 
    /* Audit */
    status = cmlAudit3( AU_MODIFY_DELAYED_RULE,  _re_id,
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone,
                        "", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlModRuleExec cmlAudit3 failure %d",
                 status );
        _rollback( "chlModRuleExec" );
        return ERROR( status, "cmlAudit3 failure" );
    }
 
 
    status =  cmlExecuteNoAnswerSql( "commit", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlModRuleExecMeta cmlExecuteNoAnswerSql commit failure %d",
                 status );
        return ERROR( status, "cmlExecuteNoAnswerSql commit failure" );
    }
 
    return CODE( status );
 
} // db_mod_rule_exec_op
 
// =-=-=-=-=-=-=-
// unregister a data object
irods::error db_del_rule_exec_op(
    irods::plugin_context& _ctx,
    const char*            _re_id ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_re_id ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    char userName[MAX_NAME_LEN + 2];
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelRuleExec" );
    }
 
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    if ( _ctx.comm()->proxyUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        if ( _ctx.comm()->proxyUser.authInfo.authFlag == LOCAL_USER_AUTH ) {
            if ( logSQL != 0 ) {
                rodsLog( LOG_SQL, "chlDelRuleExec SQL 1 " );
            }
            int status;
            {
                std::vector<std::string> bindVars;
                bindVars.push_back( _re_id );
                status = cmlGetStringValueFromSql(
                             "select user_name from R_RULE_EXEC where rule_exec_id=?",
                             userName, MAX_NAME_LEN, bindVars, &icss );
            }
            if ( status != 0 || strncmp( userName, _ctx.comm()->clientUser.userName, MAX_NAME_LEN )
                    != 0 ) {
                return ERROR( CAT_NO_ACCESS_PERMISSION, "no access permission" );
            }
        }
        else {
            return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege" );
        }
    }
 
    cllBindVars[cllBindVarCount++] = _re_id;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelRuleExec SQL 2 " );
    }
    int status =  cmlExecuteNoAnswerSql(
                      "delete from R_RULE_EXEC where rule_exec_id=?",
                      &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlDelRuleExec delete failure %d",
                 status );
        _rollback( "chlDelRuleExec" );
        return ERROR( status, "delete failure" );
    }
 
    /* Audit */
    status = cmlAudit3( AU_DELETE_DELAYED_RULE,  _re_id,
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone,
                        "", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlDelRuleExec cmlAudit3 failure %d",
                 status );
        _rollback( "chlDelRuleExec" );
        return ERROR( status, "cmlAudit3 failure" );
    }
 
    status =  cmlExecuteNoAnswerSql( "commit", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlDelRuleExec cmlExecuteNoAnswerSql commit failure %d",
                 status );
        return ERROR( status, "cmlExecuteNoAnswerSql commit failure" );
    }
 
    return CODE( status );
 
} // db_del_rule_exec_op
 
 
 
 
static irods::error extract_resource_properties_for_operations(
    const std::string& _resc_name,
    std::string& _resc_id,
    std::string& _resc_parent ) {
 
    irods::resource_ptr resc;
    irods::error ret = resc_mgr.resolve(
                           _resc_name,
                           resc);
    if(!ret.ok()) {
        return PASS(ret);
    }
 
    ret = resc->get_property<std::string>(
                    irods::RESOURCE_PARENT,
                    _resc_parent);
    if(!ret.ok()) {
        return PASS(ret);
    }
 
    rodsLong_t resc_id;
    ret = resc->get_property<rodsLong_t>(
                    irods::RESOURCE_ID,
                    resc_id);
    if(!ret.ok()) {
        return PASS(ret);
    }
 
    try {
        _resc_id = boost::lexical_cast<std::string>(resc_id);
    } catch( boost::bad_lexical_cast& ) {
        std::stringstream msg;
        msg << "failed to cast " << resc_id;
        return ERROR(
                INVALID_LEXICAL_CAST,
                msg.str());
    }
 
    return SUCCESS();
 
} // extract_resource_properties_for_operations
 
irods::error db_add_child_resc_op(
    irods::plugin_context& _ctx,
    std::map<std::string, std::string> *_resc_input ) {
    // =-=-=-=-=-=-=-
    // for readability
    std::map<std::string, std::string>& resc_input = *_resc_input;
 
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    irods::sql_logger logger( __FUNCTION__, logSQL );
 
    logger.log();
 
    std::string new_child_string( resc_input[irods::RESOURCE_CHILDREN] );
 
    irods::children_parser child_parser;
    child_parser.set_string( new_child_string );
 
    irods::children_parser::children_map_t c_map;
    child_parser.list( c_map );
 
    if(c_map.empty()) {
       return ERROR(
                  SYS_INVALID_INPUT_PARAM,
                  "child map is empty" );
    }
 
    std::string child_name    = c_map.begin()->first;
    std::string child_context = c_map.begin()->second;
 
    std::string child_resource_id;
    std::string child_parent_name;
    ret = extract_resource_properties_for_operations(
              child_name,
              child_resource_id,
              child_parent_name);
 
    if(!ret.ok()) {
    if( SYS_RESC_DOES_NOT_EXIST == ret.code() ) {
        return ERROR(
                       CHILD_NOT_FOUND,
                       child_parent_name.c_str());
    }
        return PASS(ret);
    }
 
    // check parent name, it must be empty
    if(!child_parent_name.empty()) {
        std::stringstream msg;
        msg << "Encountered an error adding '"
            << child_name
            << "' as a child resource.";
        addRErrorMsg(
            &_ctx.comm()->rError, 0,
            msg.str().c_str());
        return ERROR(
                   CHILD_HAS_PARENT,
                   msg.str() );
    }
 
 
    std::string& parent_name = resc_input[irods::RESOURCE_NAME];
    std::string parent_resource_id;
    std::string parent_parent_name;
    ret = extract_resource_properties_for_operations(
              parent_name,
              parent_resource_id,
              parent_parent_name);
    if(!ret.ok()) {
        if( SYS_RESC_DOES_NOT_EXIST == ret.code() ) {
            return ERROR(
                       CAT_INVALID_RESOURCE,
                       child_parent_name.c_str());
        }
        return PASS(ret);
    }
 
    int status = _canConnectToCatalog( _ctx.comm() );
    if(0 != status) {
        return ERROR(
                   status,
                   "_canConnectToCatalog failed");
    }
 
    std::string zone;
    ret = getLocalZone( _ctx.prop_map(), &icss, zone );
    if(!ret.ok()) {
        return PASS(ret);
    }
 
    if(resc_input[irods::RESOURCE_ZONE].length() > 0 &&
       resc_input[irods::RESOURCE_ZONE] != zone ) {
        addRErrorMsg(
            &_ctx.comm()->rError, 0,
            "Currently, resources must be in the local zone" );
 
        return ERROR(
                   CAT_INVALID_ZONE,
                   "resources must be in the local zone");
    }
 
    logger.log();
 
    ret = _updateChildParent(
              child_resource_id,
              parent_resource_id,
              child_context );
    if(!ret.ok()) {
        return PASS(ret);
    }
 
    status = cmlExecuteNoAnswerSql( "commit", &icss );
    if(status != 0) {
        return ERROR(
                   status,
                   "commit failure");
    }
 
    return SUCCESS();
 
} // db_add_child_resc_op
 
// =-=-=-=-=-=-=-
//
irods::error db_reg_resc_op(
    irods::plugin_context& _ctx,
    std::map<std::string, std::string> *_resc_input ) {
 
    // =-=-=-=-=-=-=-
    // check the params
    if ( !_resc_input ) {
        return ERROR( SYS_INTERNAL_NULL_INPUT_ERR, "NULL parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // for readability
    std::map<std::string, std::string>& resc_input = *_resc_input;
 
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    rodsLong_t seqNum;
    char idNum[MAX_SQL_SIZE];
    int status;
    char myTime[50];
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegResc" );
    }
 
    // =-=-=-=-=-=-=-
    // error trap empty resc name
    if ( resc_input[irods::RESOURCE_NAME].length() < 1 ) {
        addRErrorMsg( &_ctx.comm()->rError, 0, "resource name is empty" );
        return ERROR( CAT_INVALID_RESOURCE_NAME, "resource name is empty" );
    }
 
    // =-=-=-=-=-=-=-
    // error trap empty resc type
    if ( resc_input[irods::RESOURCE_TYPE].length() < 1 ) {
        addRErrorMsg( &_ctx.comm()->rError, 0, "resource type is empty" );
        return ERROR( CAT_INVALID_RESOURCE_TYPE, "resource type is empty" );
    }
 
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    if ( _ctx.comm()->clientUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege" );
    }
    if ( _ctx.comm()->proxyUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege level" );
    }
 
    // =-=-=-=-=-=-=-
    // Validate resource name format
    ret = validate_resource_name( resc_input[irods::RESOURCE_NAME] );
    if ( !ret.ok() ) {
        irods::log( ret );
        return PASS( ret );
    }
    // =-=-=-=-=-=-=-
 
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegResc SQL 1 " );
    }
    seqNum = cmlGetNextSeqVal( &icss );
    if ( seqNum < 0 ) {
        rodsLog( LOG_NOTICE, "chlRegResc cmlGetNextSeqVal failure %d",
                 seqNum );
        _rollback( "chlRegResc" );
        return ERROR( seqNum, "cmlGetNextSeqVal failure" );
    }
    snprintf( idNum, MAX_SQL_SIZE, "%lld", seqNum );
 
    std::string zone;
    ret = getLocalZone( _ctx.prop_map(), &icss, zone );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }
 
    if ( resc_input[irods::RESOURCE_ZONE].length() > 0 ) {
        if ( resc_input[irods::RESOURCE_ZONE] != zone ) {
            addRErrorMsg( &_ctx.comm()->rError, 0,
                          "Currently, resources must be in the local zone" );
            return ERROR( CAT_INVALID_ZONE, "resources must be in the local zone" );
        }
    }
    // =-=-=-=-=-=-=-
    // JMC :: resources may now have an empty location if they
    //     :: are coordinating nodes
    //    if (strlen(_resc_info->rescLoc)<1) {
    //        return(CAT_INVALID_RESOURCE_NET_ADDR);
    //    }
    // =-=-=-=-=-=-=-
    // if the resource is not the 'empty resource' test it
    if ( resc_input[irods::RESOURCE_LOCATION] != irods::EMPTY_RESC_HOST ) {
        // =-=-=-=-=-=-=-
        // JMC - backport 4597
        _resolveHostName( _ctx.comm(), resc_input[irods::RESOURCE_LOCATION].c_str());
    }
 
    // Root dir is not a valid vault path
    ret = verify_non_root_vault_path(_ctx, resc_input[irods::RESOURCE_PATH]);
    if (!ret.ok()) {
        return PASS(ret);
    }
 
    getNowStr( myTime );
 
    cllBindVars[0] = idNum;
    cllBindVars[1] = resc_input[irods::RESOURCE_NAME].c_str();
    cllBindVars[2] = ( char* )zone.c_str();
    cllBindVars[3] = resc_input[irods::RESOURCE_TYPE].c_str();
    cllBindVars[4] = resc_input[irods::RESOURCE_CLASS].c_str();
    cllBindVars[5] = resc_input[irods::RESOURCE_LOCATION].c_str();
    cllBindVars[6] = resc_input[irods::RESOURCE_PATH].c_str();
    cllBindVars[7] = myTime;
    cllBindVars[8] = myTime;
    cllBindVars[9] = resc_input[irods::RESOURCE_CHILDREN].c_str();
    cllBindVars[10] = resc_input[irods::RESOURCE_CONTEXT].c_str();
    cllBindVars[11] = resc_input[irods::RESOURCE_PARENT].c_str();
    cllBindVarCount = 12;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegResc SQL 4" );
    }
    status =  cmlExecuteNoAnswerSql(
                  "insert into R_RESC_MAIN (resc_id, resc_name, zone_name, resc_type_name, resc_class_name, resc_net, resc_def_path, create_ts, modify_ts, resc_children, resc_context, resc_parent) values (?,?,?,?,?,?,?,?,?,?,?,?)",
                  &icss );
 
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegResc cmlExectuteNoAnswerSql(insert) failure %d",
                 status );
        _rollback( "chlRegResc" );
        return ERROR( status, "cmlExectuteNoAnswerSql(insert) failure" );
    }
 
    /* Audit */
    status = cmlAudit3( AU_REGISTER_RESOURCE,  idNum,
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone,
                        resc_input[irods::RESOURCE_NAME].c_str(), &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegResc cmlAudit3 failure %d",
                 status );
        _rollback( "chlRegResc" );
        return ERROR( status, "chlRegResc cmlAudit3 failure" );
    }
 
    status =  cmlExecuteNoAnswerSql( "commit", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegResc cmlExecuteNoAnswerSql commit failure %d", status );
        return ERROR( status, "cmlExecuteNoAnswerSql commit failure" );
    }
 
    return CODE( status );
 
} // db_reg_resc_op
 
// =-=-=-=-=-=-=-
//
irods::error db_del_child_resc_op(
    irods::plugin_context& _ctx,
    std::map<std::string, std::string> *_resc_input ) {
 
    // =-=-=-=-=-=-=-
    // check the params
    if ( !_resc_input ) {
        return ERROR( SYS_INTERNAL_NULL_INPUT_ERR, "NULL parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // for readability
    std::map<std::string, std::string>& resc_input = *_resc_input;
 
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    irods::sql_logger logger( "chlDelChildResc", logSQL );
    std::string child_string( resc_input[irods::RESOURCE_CHILDREN] );
 
    std::string& parent_name = resc_input[irods::RESOURCE_NAME];
    std::string parent_resource_id;
    std::string parent_parent_resource_id;
    ret = extract_resource_properties_for_operations(
              parent_name,
              parent_resource_id,
              parent_parent_resource_id);
    if(!ret.ok()) {
           return PASS(ret);
    }
 
    irods::children_parser parser;
    parser.set_string( child_string );
 
    std::string child_name;
    parser.first_child( child_name );
 
    std::string child_resource_id;
    std::string child_parent_resource_id;
    ret = extract_resource_properties_for_operations(
              child_name,
              child_resource_id,
              child_parent_resource_id);
    if(!ret.ok()) {
           return PASS(ret);
    }
 
    if (child_parent_resource_id != parent_resource_id) {
        return ERROR(CAT_INVALID_CHILD, "invalid parent/child relationship");
    }
 
    int status = _canConnectToCatalog( _ctx.comm() );
    if(0 != status) {
        return ERROR(
                   status,
                   "_canConnectToCatalog failed");
    }
 
    std::string zone;
    ret = getLocalZone( _ctx.prop_map(), &icss, zone );
    if(!ret.ok()) {
        return PASS(ret);
    }
 
    ret = _updateChildParent(
              child_resource_id,
              std::string(""),
              std::string(""));
    if(!ret.ok()) {
        return PASS(ret);
    }
 
    status = cmlExecuteNoAnswerSql( "commit", &icss );
    if(status != 0) {
        return ERROR(
                   status,
                   "commit failure");
    }
 
    return SUCCESS();
 
} // db_del_child_resc_op
 
// =-=-=-=-=-=-=-
// delete a resource
irods::error db_del_resc_op(
    irods::plugin_context& _ctx,
    const char *_resc_name,
    int                    _dry_run ) {
 
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if ( !_resc_name ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
 
    int status;
    char rescId[MAX_NAME_LEN];
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelResc" );
    }
 
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    if ( _ctx.comm()->clientUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege" );
    }
    if ( _ctx.comm()->proxyUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege" );
    }
 
    // =-=-=-=-=-=-=-
    // JMC - backport 4629
    if ( strncmp( _resc_name, BUNDLE_RESC, strlen( BUNDLE_RESC ) ) == 0 ) {
        char errMsg[155];
        snprintf( errMsg, 150,
                  "%s is a built-in resource needed for bundle operations.",
                  BUNDLE_RESC );
        addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
        return ERROR( CAT_PSEUDO_RESC_MODIFY_DISALLOWED, "cannot delete bundle resc" );
    }
    // =-=-=-=-=-=-=-
 
    bool has_data = true; // default to error case
    status = _rescHasData( &icss, _resc_name, has_data );
    if( status < 0 ) {
        rodsLog(
            LOG_ERROR,
            "%s - _rescHasData failed for [%s] %d",
            __FUNCTION__,
            _resc_name,
            status );
        return ERROR(
                  status,
                  "failed to get object count for resource" );
    }
 
    if( has_data   ) {
        char errMsg[105];
        snprintf( errMsg, 100,
                  "resource '%s' contains one or more dataObjects",
                  _resc_name );
        addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
        return ERROR( CAT_RESOURCE_NOT_EMPTY, "resc not empty" );
    }
 
    std::string zone;
    ret = getLocalZone( _ctx.prop_map(), &icss, zone );
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    /* get rescId for possible audit call; won't be available after delete */
    rescId[0] = '\0';
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelResc SQL 2 " );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( _resc_name );
        status = cmlGetStringValueFromSql(
                     "select resc_id from R_RESC_MAIN where resc_name=?",
                     rescId, MAX_NAME_LEN, bindVars, &icss );
    }
    if ( status != 0 ) {
        if ( status == CAT_SUCCESS_BUT_WITH_NO_INFO ) {
            char errMsg[105];
            snprintf( errMsg, 100,
                      "resource '%s' does not exist",
                      _resc_name );
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
            return ERROR( status, "resource does not exits" );
        }
        _rollback( "chlDelResc" );
        return ERROR( status, "resource does not exist" );
    }
 
    if ( _rescHasParentOrChild( rescId ) ) {
        char errMsg[105];
        snprintf( errMsg, 100,
                  "resource '%s' has a parent or child",
                  _resc_name );
        addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
        return ERROR( CAT_RESOURCE_NOT_EMPTY, "resource not empty" );
    }
 
    cllBindVars[cllBindVarCount++] = _resc_name;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelResc SQL 3" );
    }
    status = cmlExecuteNoAnswerSql(
                 "delete from R_RESC_MAIN where resc_name=?",
                 &icss );
    if ( status != 0 ) {
        if ( status == CAT_SUCCESS_BUT_WITH_NO_INFO ) {
            char errMsg[105];
            snprintf( errMsg, 100,
                      "resource '%s' does not exist",
                      _resc_name );
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
            return ERROR( status, "resource does not exist" );
        }
        _rollback( "chlDelResc" );
        return ERROR( status, "resource does not exist" );
    }
 
    /* Remove associated AVUs, if any */
    removeMetaMapAndAVU( rescId );
 
 
    /* Audit */
    status = cmlAudit3( AU_DELETE_RESOURCE,
                        rescId,
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone,
                        _resc_name,
                        &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlDelResc cmlAudit3 failure %d",
                 status );
        _rollback( "chlDelResc" );
        return ERROR( status, "cmlAudi3 failure" );
    }
 
    if ( _dry_run ) { // JMC
        _rollback( "chlDelResc" );
        return CODE( status );
    }
 
    status =  cmlExecuteNoAnswerSql( "commit", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlDelResc cmlExecuteNoAnswerSql commit failure %d",
                 status );
        return ERROR( status, "cmlExecuteNoAnswerSql commit failure" );
    }
    return CODE( status );
 
} // db_del_resc_op
 
// =-=-=-=-=-=-=-
// rollback the db
irods::error db_rollback_op(
    irods::plugin_context& _ctx ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRollback - SQL 1 " );
    }
 
    int status =  cmlExecuteNoAnswerSql( "rollback", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRollback cmlExecuteNoAnswerSql failure %d",
                 status );
        return ERROR( status, "chlRollback cmlExecuteNoAnswerSql failure" );
    }
 
    return CODE( status );
 
} // db_rollback_op
 
// =-=-=-=-=-=-=-
// commit the transaction
irods::error db_commit_op(
    irods::plugin_context& _ctx ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlCommit - SQL 1 " );
    }
    int status =  cmlExecuteNoAnswerSql( "commit", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlCommit cmlExecuteNoAnswerSql failure %d",
                 status );
        return ERROR( status, "chlCommit cmlExecuteNoAnswerSql failure" );
    }
 
    return CODE( status );
 
} // db_commit_op
 
// =-=-=-=-=-=-=-
// commit the transaction
irods::error db_del_user_re_op(
    irods::plugin_context& _ctx,
    userInfo_t*            _user_info ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_user_info ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    int status;
    char iValStr[200];
    char zoneToUse[MAX_NAME_LEN];
    char userStr[200];
    char userName2[NAME_LEN];
    char zoneName[NAME_LEN];
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelUserRE" );
    }
 
    if ( _ctx.comm()->clientUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege" );
    }
    if ( _ctx.comm()->proxyUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege level" );
    }
 
    std::string zone;
    ret = getLocalZone( _ctx.prop_map(), &icss, zone );
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    snprintf( zoneToUse, sizeof( zoneToUse ), "%s", zone.c_str() );
    if ( strlen( _user_info->rodsZone ) > 0 ) {
        snprintf( zoneToUse, sizeof( zoneToUse ), "%s", _user_info->rodsZone );
    }
 
    status = validateAndParseUserName( _user_info->userName, userName2, zoneName );
    if ( status ) {
        return ERROR( status, "Invalid username format" );
    }
    if ( zoneName[0] != '\0' ) {
        rstrcpy( zoneToUse, zoneName, NAME_LEN );
    }
 
    if ( strncmp( _ctx.comm()->clientUser.userName, userName2, sizeof( userName2 ) ) == 0 &&
            strncmp( _ctx.comm()->clientUser.rodsZone, zoneToUse, sizeof( zoneToUse ) ) == 0 ) {
        addRErrorMsg( &_ctx.comm()->rError, 0, "Cannot remove your own admin account, probably unintended" );
        return ERROR( CAT_INVALID_USER, "invalid user" );
    }
 
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelUserRE SQL 1 " );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( userName2 );
        bindVars.push_back( zoneToUse );
        status = cmlGetStringValueFromSql(
                     "select user_id from R_USER_MAIN where user_name=? and zone_name=?",
                     iValStr, 200, bindVars, &icss );
    }
    if ( status == CAT_SUCCESS_BUT_WITH_NO_INFO ||
            status == CAT_NO_ROWS_FOUND ) {
        addRErrorMsg( &_ctx.comm()->rError, 0, "Invalid user" );
        return ERROR( CAT_INVALID_USER, "invalid user" );
    }
    if ( status != 0 ) {
        _rollback( "chlDelUserRE" );
        return ERROR( status, "failed getting user from table" );
    }
 
    cllBindVars[cllBindVarCount++] = userName2;
    cllBindVars[cllBindVarCount++] = zoneToUse;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelUserRE SQL 2" );
    }
    status = cmlExecuteNoAnswerSql(
                 "delete from R_USER_MAIN where user_name=? and zone_name=?",
                 &icss );
    if ( status == CAT_SUCCESS_BUT_WITH_NO_INFO ) {
        return ERROR( CAT_INVALID_USER, "invalid user" );
    }
    if ( status != 0 ) {
        _rollback( "chlDelUserRE" );
        return ERROR( status, "cmlExecuteNoAnswerSql for delete user failed" );
    }
 
    cllBindVars[cllBindVarCount++] = iValStr;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelUserRE SQL 3" );
    }
    status = cmlExecuteNoAnswerSql(
                 "delete from R_USER_PASSWORD where user_id=?",
                 &icss );
    if ( status != 0 && status != CAT_SUCCESS_BUT_WITH_NO_INFO ) {
        char errMsg[MAX_NAME_LEN + 40];
        rodsLog( LOG_NOTICE,
                 "chlDelUserRE delete password failure %d",
                 status );
        snprintf( errMsg, sizeof errMsg, "Error removing password entry" );
        addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
        _rollback( "chlDelUserRE" );
        return ERROR( status, "Error removing password entry" );
    }
 
    /* Remove both the special user_id = group_user_id entry and any
       other access entries for this user (or group) */
    cllBindVars[cllBindVarCount++] = iValStr;
    cllBindVars[cllBindVarCount++] = iValStr;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelUserRE SQL 4" );
    }
    status = cmlExecuteNoAnswerSql(
                 "delete from R_USER_GROUP where user_id=? or group_user_id=?",
                 &icss );
    if ( status != 0 && status != CAT_SUCCESS_BUT_WITH_NO_INFO ) {
        char errMsg[MAX_NAME_LEN + 40];
        rodsLog( LOG_NOTICE,
                 "chlDelUserRE delete user_group entry failure %d",
                 status );
        snprintf( errMsg, sizeof errMsg, "Error removing user_group entry" );
        addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
        _rollback( "chlDelUserRE" );
        return ERROR( status, "Error removing user_group entry" );
    }
 
    /* Remove any R_USER_AUTH rows for this user */
    cllBindVars[cllBindVarCount++] = iValStr;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlDelUserRE SQL 4" );
    }
    status = cmlExecuteNoAnswerSql(
                 "delete from R_USER_AUTH where user_id=?",
                 &icss );
    if ( status != 0 && status != CAT_SUCCESS_BUT_WITH_NO_INFO ) {
        char errMsg[MAX_NAME_LEN + 40];
        rodsLog( LOG_NOTICE,
                 "chlDelUserRE delete user_auth entries failure %d",
                 status );
        snprintf( errMsg, sizeof errMsg, "Error removing user_auth entries" );
        addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
        _rollback( "chlDelUserRE" );
        return ERROR( status, "Error removing user_auth entries" );
    }
 
    /* Remove associated AVUs, if any */
    removeMetaMapAndAVU( iValStr );
 
    /* Audit */
    snprintf( userStr, sizeof userStr, "%s#%s",
              userName2, zoneToUse );
    status = cmlAudit3( AU_DELETE_USER_RE,
                        iValStr,
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone,
                        userStr,
                        &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlDelUserRE cmlAudit3 failure %d",
                 status );
        _rollback( "chlDelUserRE" );
        return ERROR( status, "chlDelUserRE cmlAudit3 failure" );
    }
 
    return SUCCESS();
 
} // db_del_user_re_op
 
// =-=-=-=-=-=-=-
// commit the transaction
irods::error db_reg_coll_by_admin_op(
    irods::plugin_context& _ctx,
    collInfo_t*            _coll_info ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_coll_info ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    char myTime[50];
    char logicalEndName[MAX_NAME_LEN];
    char logicalParentDirName[MAX_NAME_LEN];
    rodsLong_t iVal;
    char collIdNum[MAX_NAME_LEN];
    char nextStr[MAX_NAME_LEN];
    char currStr[MAX_NAME_LEN];
    char currStr2[MAX_SQL_SIZE];
    int status;
    char tSQL[MAX_SQL_SIZE];
    char userName2[NAME_LEN];
    char zoneName[NAME_LEN];
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegCollByAdmin" );
    }
 
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    // =-=-=-=-=-=-=-
    // JMC - backport 4772
    if ( _ctx.comm()->clientUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ||
            _ctx.comm()->proxyUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        int status2;
        status2  = cmlCheckGroupAdminAccess(
                       _ctx.comm()->clientUser.userName,
                       _ctx.comm()->clientUser.rodsZone,
                       "", &icss );
        if ( status2 != 0 ) {
            return ERROR( status2, "no group admin access" );
        }
        if ( creatingUserByGroupAdmin == 0 ) {
            return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficent privilege" );
        }
        // =-=-=-=-=-=-=-
    }
 
    status = splitPathByKey( _coll_info->collName,
                             logicalParentDirName, MAX_NAME_LEN, logicalEndName, MAX_NAME_LEN, '/' );
 
    if ( strlen( logicalParentDirName ) == 0 ) {
        snprintf( logicalParentDirName, sizeof( logicalParentDirName ), "%s", PATH_SEPARATOR );
        snprintf( logicalEndName, sizeof( logicalEndName ), "%s", _coll_info->collName + 1 );
    }
 
    /* Check that the parent collection exists */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegCollByAdmin SQL 1 " );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( logicalParentDirName );
        status = cmlGetIntegerValueFromSql(
                     "select coll_id from R_COLL_MAIN where coll_name=?",
                     &iVal, bindVars, &icss );
    }
    if ( status < 0 ) {
        char errMsg[MAX_NAME_LEN + 40];
        if ( status == CAT_NO_ROWS_FOUND ) {
            snprintf( errMsg, sizeof errMsg,
                      "collection '%s' is unknown, cannot create %s under it",
                      logicalParentDirName, logicalEndName );
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
            return ERROR( status, "collection is unknown" );
        }
        _rollback( "chlRegCollByAdmin" );
        return ERROR( status, "collection not found" );
    }
 
    snprintf( collIdNum, MAX_NAME_LEN, "%d", status );
 
    /* String to get next sequence item for objects */
    cllNextValueString( "R_ObjectID", nextStr, MAX_NAME_LEN );
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegCollByAdmin SQL 2" );
    }
    snprintf( tSQL, MAX_SQL_SIZE,
              "insert into R_COLL_MAIN (coll_id, parent_coll_name, coll_name, coll_owner_name, coll_owner_zone, coll_type, coll_info1, coll_info2, create_ts, modify_ts) values (%s, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
              nextStr );
 
    getNowStr( myTime );
 
    std::string zone;
    ret = getLocalZone( _ctx.prop_map(), &icss, zone );
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    /* Parse input name into user and zone */
    status = validateAndParseUserName( _coll_info->collOwnerName, userName2, zoneName );
    if ( status ) {
        return ERROR( status, "Invalid username format" );
    }
    if ( zoneName[0] == '\0' ) {
        rstrcpy( zoneName, zone.c_str(), NAME_LEN );
    }
 
    cllBindVars[cllBindVarCount++] = logicalParentDirName;
    cllBindVars[cllBindVarCount++] = _coll_info->collName;
    cllBindVars[cllBindVarCount++] = userName2;
    if ( strlen( _coll_info->collOwnerZone ) > 0 ) {
        cllBindVars[cllBindVarCount++] = _coll_info->collOwnerZone;
    }
    else {
        cllBindVars[cllBindVarCount++] = zoneName;
    }
    cllBindVars[cllBindVarCount++] = _coll_info->collType;
    cllBindVars[cllBindVarCount++] = _coll_info->collInfo1;
    cllBindVars[cllBindVarCount++] = _coll_info->collInfo2;
    cllBindVars[cllBindVarCount++] = myTime;
    cllBindVars[cllBindVarCount++] = myTime;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegCollByAdmin SQL 3" );
    }
    status =  cmlExecuteNoAnswerSql( tSQL,
                                     &icss );
    if ( status != 0 ) {
        char errMsg[105];
        if ( status == CATALOG_ALREADY_HAS_ITEM_BY_THAT_NAME ) {
            snprintf( errMsg, 100, "Error %d %s",
                      status,
                      "CATALOG_ALREADY_HAS_ITEM_BY_THAT_NAME"
                    );
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
        }
 
        rodsLog( LOG_NOTICE,
                 "chlRegCollByAdmin cmlExecuteNoAnswerSQL(insert) failure %d"
                 , status );
        _rollback( "chlRegCollByAdmin" );
        return ERROR( status, "cmlExecuteNoAnswerSQL(insert) failure" );
    }
 
    /* String to get current sequence item for objects */
    cllCurrentValueString( "R_ObjectID", currStr, MAX_NAME_LEN );
    snprintf( currStr2, MAX_SQL_SIZE, " %s ", currStr );
 
    cllBindVars[cllBindVarCount++] = userName2;
    cllBindVars[cllBindVarCount++] = zoneName;
    cllBindVars[cllBindVarCount++] = ACCESS_OWN;
    cllBindVars[cllBindVarCount++] = myTime;
    cllBindVars[cllBindVarCount++] = myTime;
 
    snprintf( tSQL, MAX_SQL_SIZE,
              "insert into R_OBJT_ACCESS values (%s, (select user_id from R_USER_MAIN where user_name=? and zone_name=?), (select token_id from R_TOKN_MAIN where token_namespace = 'access_type' and token_name = ?), ?, ?)",
              currStr2 );
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegCollByAdmin SQL 4" );
    }
    status =  cmlExecuteNoAnswerSql( tSQL, &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegCollByAdmin cmlExecuteNoAnswerSql(insert access) failure %d",
                 status );
        _rollback( "chlRegCollByAdmin" );
        return ERROR( status, "cmlExecuteNoAnswerSql(insert access) failure" );
    }
 
    /* Audit */
    status = cmlAudit4( AU_REGISTER_COLL_BY_ADMIN,
                        currStr2,
                        "",
                        userName2,
                        zoneName,
                        _ctx.comm()->clientUser.userName,
                        &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegCollByAdmin cmlAudit4 failure %d",
                 status );
        _rollback( "chlRegCollByAdmin" );
        return ERROR( status, "cmlAudit4 failure" );
    }
 
    return SUCCESS();
 
} // db_reg_coll_by_admin_op
 
// =-=-=-=-=-=-=-
// commit the transaction
irods::error db_reg_coll_op(
    irods::plugin_context& _ctx,
    collInfo_t*            _coll_info ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_coll_info ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    char myTime[50];
    char logicalEndName[MAX_NAME_LEN];
    char logicalParentDirName[MAX_NAME_LEN];
    rodsLong_t iVal;
    char collIdNum[MAX_NAME_LEN];
    char nextStr[MAX_NAME_LEN];
    char currStr[MAX_NAME_LEN];
    char currStr2[MAX_SQL_SIZE];
    rodsLong_t status;
    char tSQL[MAX_SQL_SIZE];
    int inheritFlag;
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegColl" );
    }
 
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    status = splitPathByKey( _coll_info->collName,
                             logicalParentDirName, MAX_NAME_LEN, logicalEndName, MAX_NAME_LEN, '/' );
 
    if ( strlen( logicalParentDirName ) == 0 ) {
        snprintf( logicalParentDirName, sizeof( logicalParentDirName ), "%s", PATH_SEPARATOR );
        snprintf( logicalEndName, sizeof( logicalEndName ), "%s", _coll_info->collName + 1 );
    }
 
    /* Check that the parent collection exists and user has write permission,
       and get the collectionID.  Also get the inherit flag */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegColl SQL 1 " );
    }
 
    status = cmlCheckDirAndGetInheritFlag( logicalParentDirName,
                                           _ctx.comm()->clientUser.userName,
                                           _ctx.comm()->clientUser.rodsZone,
                                           ACCESS_MODIFY_OBJECT, &inheritFlag,
                                           mySessionTicket, mySessionClientAddr, &icss );
    if ( status < 0 ) {
        char errMsg[105];
        if ( status == CAT_UNKNOWN_COLLECTION ) {
            snprintf( errMsg, 100, "collection '%s' is unknown",
                      logicalParentDirName );
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg );
            return ERROR( status, "collection is unknown" );
        }
        _rollback( "chlRegColl" );
        return ERROR( status, "cmlCheckDirAndGetInheritFlag failed" );
    }
    snprintf( collIdNum, MAX_NAME_LEN, "%lld", status );
 
    /* Check that the path is not already a dataObj */
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegColl SQL 2" );
    }
    {
        std::vector<std::string> bindVars;
        bindVars.push_back( logicalEndName );
        bindVars.push_back( collIdNum );
        status = cmlGetIntegerValueFromSql(
                     "select data_id from R_DATA_MAIN where data_name=? and coll_id=?",
                     &iVal, bindVars, &icss );
    }
 
    if ( status == 0 ) {
        return ERROR( CAT_NAME_EXISTS_AS_DATAOBJ, "data obj alread exists" );
    }
 
 
    /* String to get next sequence item for objects */
    cllNextValueString( "R_ObjectID", nextStr, MAX_NAME_LEN );
 
    getNowStr( myTime );
 
    cllBindVars[cllBindVarCount++] = logicalParentDirName;
    cllBindVars[cllBindVarCount++] = _coll_info->collName;
    cllBindVars[cllBindVarCount++] = _ctx.comm()->clientUser.userName;
    cllBindVars[cllBindVarCount++] = _ctx.comm()->clientUser.rodsZone;
    cllBindVars[cllBindVarCount++] = _coll_info->collType;
    cllBindVars[cllBindVarCount++] = _coll_info->collInfo1;
    cllBindVars[cllBindVarCount++] = _coll_info->collInfo2;
    cllBindVars[cllBindVarCount++] = myTime;
    cllBindVars[cllBindVarCount++] = myTime;
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegColl SQL 3" );
    }
    snprintf( tSQL, MAX_SQL_SIZE,
              "insert into R_COLL_MAIN (coll_id, parent_coll_name, coll_name, coll_owner_name, coll_owner_zone, coll_type, coll_info1, coll_info2, create_ts, modify_ts) values (%s, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
              nextStr );
    status =  cmlExecuteNoAnswerSql( tSQL,
                                     &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegColl cmlExecuteNoAnswerSql(insert) failure %d", status );
        _rollback( "chlRegColl" );
        return ERROR( status, "cmlExecuteNoAnswerSql(insert) failure" );
    }
 
    /* String to get current sequence item for objects */
    cllCurrentValueString( "R_ObjectID", currStr, MAX_NAME_LEN );
    snprintf( currStr2, MAX_SQL_SIZE, " %s ", currStr );
 
    if ( inheritFlag ) {
        /* If inherit is set (sticky bit), then add access rows for this
           collection that match those of the parent collection */
        cllBindVars[0] = myTime;
        cllBindVars[1] = myTime;
        cllBindVars[2] = collIdNum;
        cllBindVarCount = 3;
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlRegColl SQL 4" );
        }
        snprintf( tSQL, MAX_SQL_SIZE,
                  "insert into R_OBJT_ACCESS (object_id, user_id, access_type_id, create_ts, modify_ts) (select %s, user_id, access_type_id, ?, ? from R_OBJT_ACCESS where object_id = ?)",
                  currStr2 );
        status =  cmlExecuteNoAnswerSql( tSQL, &icss );
 
        if ( status == 0 ) {
            if ( logSQL != 0 ) {
                rodsLog( LOG_SQL, "chlRegColl SQL 5" );
            }
#if ORA_ICAT
            char newCollectionID[MAX_NAME_LEN];
            /*
               For Oracle, we can't use currStr2 string in a where clause so
               do another query to get the new collection id.
            */
            status = cmlGetCurrentSeqVal( &icss );
 
            if ( status > 0 ) {
                /* And then use it in the where clause for the update */
                snprintf( newCollectionID, MAX_NAME_LEN, "%lld", status );
                cllBindVars[cllBindVarCount++] = "1";
                cllBindVars[cllBindVarCount++] = myTime;
                cllBindVars[cllBindVarCount++] = newCollectionID;
                status =  cmlExecuteNoAnswerSql(
                              "update R_COLL_MAIN set coll_inheritance=?, modify_ts=? where coll_id=?",
                              &icss );
            }
#else
            /*
              For Postgres we can, use the currStr2 to get the current id
              and save a SQL interaction.
            */
            cllBindVars[cllBindVarCount++] = "1";
            cllBindVars[cllBindVarCount++] = myTime;
            snprintf( tSQL, MAX_SQL_SIZE,
                      "update R_COLL_MAIN set coll_inheritance=?, modify_ts=? where coll_id=%s",
                      currStr2 );
            status =  cmlExecuteNoAnswerSql( tSQL, &icss );
#endif
        }
    }
    else {
        cllBindVars[cllBindVarCount++] = _ctx.comm()->clientUser.userName;
        cllBindVars[cllBindVarCount++] = _ctx.comm()->clientUser.rodsZone;
        cllBindVars[cllBindVarCount++] = ACCESS_OWN;
        cllBindVars[cllBindVarCount++] = myTime;
        cllBindVars[cllBindVarCount++] = myTime;
        snprintf( tSQL, MAX_SQL_SIZE,
                  "insert into R_OBJT_ACCESS values (%s, (select user_id from R_USER_MAIN where user_name=? and zone_name=?), (select token_id from R_TOKN_MAIN where token_namespace = 'access_type' and token_name = ?), ?, ?)",
                  currStr2 );
        if ( logSQL != 0 ) {
            rodsLog( LOG_SQL, "chlRegColl SQL 6" );
        }
        status =  cmlExecuteNoAnswerSql( tSQL, &icss );
    }
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegColl cmlExecuteNoAnswerSql(insert access) failure %d",
                 status );
        _rollback( "chlRegColl" );
        return ERROR( status, "cmlExecuteNoAnswerSql(insert access) failure" );
    }
 
    /* Audit */
    status = cmlAudit4( AU_REGISTER_COLL,
                        currStr2,
                        "",
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone,
                        _coll_info->collName,
                        &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegColl cmlAudit4 failure %d",
                 status );
        _rollback( "chlRegColl" );
        return ERROR( status, "cmlAudit4 failure" );
    }
 
    status =  cmlExecuteNoAnswerSql( "commit", &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlRegColl cmlExecuteNoAnswerSql commit failure %d",
                 status );
        return ERROR( status, "cmlExecuteNoAnswerSql commit failure" );
    }
 
    return CODE( status );
 
} // db_reg_coll_op
 
// =-=-=-=-=-=-=-
// commit the transaction
irods::error db_mod_coll_op(
    irods::plugin_context& _ctx,
    collInfo_t*            _coll_info ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_coll_info ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    char myTime[50];
    rodsLong_t status;
    int count;
    rodsLong_t iVal;
    char iValStr[60];
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlModColl" );
    }
 
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    /* Check that collection exists and user has write permission */
    iVal = cmlCheckDir( _coll_info->collName,  _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone,
                        ACCESS_MODIFY_OBJECT, &icss );
 
    if ( iVal < 0 ) {
        if ( iVal == CAT_UNKNOWN_COLLECTION ) {
            std::stringstream errMsg;
            errMsg << "collection '" << _coll_info->collName << "' is unknown";
            addRErrorMsg( &_ctx.comm()->rError, 0, errMsg.str().c_str() );
            return ERROR( CAT_UNKNOWN_COLLECTION, "unknown collection" );
        }
 
        if ( iVal == CAT_NO_ACCESS_PERMISSION ) {
            // Allows elevation of privileges (e.g. irods_rule_engine_plugin-update_collection_mtime).
            if (irods::is_privileged_client(*_ctx.comm())) {
                iVal = 0;
            }
            else {
                std::stringstream errMsg;
                errMsg << "no permission to update collection '" << _coll_info->collName << "'";
                addRErrorMsg( &_ctx.comm()->rError, 0, errMsg.str().c_str() );
                return ERROR( CAT_NO_ACCESS_PERMISSION, "no permission" );
            }
        }
 
        // If client privileges are elevated, then iVal must be checked again because
        // it could have been modified (e.g. irods_rule_engine_plugin-update_collection_mtime).
        if (iVal < 0) {
            return ERROR( iVal, "cmlCheckDir failed" );
        }
    }
 
    std::string tSQL( "update R_COLL_MAIN set " );
    count = 0;
 
    if ( strlen( _coll_info->collType ) > 0 ) {
        if ( strcmp( _coll_info->collType, "NULL_SPECIAL_VALUE" ) == 0 ) {
            /* A special value to indicate NULL */
            cllBindVars[cllBindVarCount++] = "";
        }
        else {
            cllBindVars[cllBindVarCount++] = _coll_info->collType;
        }
        tSQL += "coll_type=? ";
        count++;
    }
 
    if ( strlen( _coll_info->collInfo1 ) > 0 ) {
        if ( strcmp( _coll_info->collInfo1, "NULL_SPECIAL_VALUE" ) == 0 ) {
            /* A special value to indicate NULL */
            cllBindVars[cllBindVarCount++] = "";
        }
        else {
            cllBindVars[cllBindVarCount++] = _coll_info->collInfo1;
        }
        if ( count > 0 ) {
            tSQL += ",";
        }
        tSQL += "coll_info1=? ";
        count++;
    }
 
    if ( strlen( _coll_info->collInfo2 ) > 0 ) {
        if ( strcmp( _coll_info->collInfo2, "NULL_SPECIAL_VALUE" ) == 0 ) {
            /* A special value to indicate NULL */
            cllBindVars[cllBindVarCount++] = "";
        }
        else {
            cllBindVars[cllBindVarCount++] = _coll_info->collInfo2;
        }
        if ( count > 0 ) {
            tSQL += ",";
        }
        tSQL += "coll_info2=? ";
        count++;
    }
 
    if (strlen(_coll_info->collModify) > 0) {
        cllBindVars[cllBindVarCount++] = _coll_info->collModify;
 
        if (count > 0) {
            tSQL += ',';
        }
 
        ++count;
    }
    else {
        tSQL += ',';
        getNowStr( myTime );
        cllBindVars[cllBindVarCount++] = myTime;
    }
 
    if ( count == 0 ) {
        return ERROR( CAT_INVALID_ARGUMENT, "count is 0" );
    }
 
    cllBindVars[cllBindVarCount++] = _coll_info->collName;
    tSQL += " modify_ts=? where coll_name=?";
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlModColl SQL 1" );
    }
    status =  cmlExecuteNoAnswerSql( tSQL.c_str(),
                                     &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlModColl cmlExecuteNoAnswerSQL(update) failure %d", status );
        return ERROR( status, "cmlExecuteNoAnswerSQL(update) failure" );
    }
 
    /* Audit */
    snprintf( iValStr, sizeof iValStr, "%lld", iVal );
    status = cmlAudit3( AU_REGISTER_COLL,
                        iValStr,
                        _ctx.comm()->clientUser.userName,
                        _ctx.comm()->clientUser.rodsZone,
                        _coll_info->collName,
                        &icss );
    if ( status != 0 ) {
        rodsLog( LOG_NOTICE,
                 "chlModColl cmlAudit3 failure %d",
                 status );
        return ERROR( status, "cmlAudit3 failure" );
    }
 
    return SUCCESS();
 
} // db_mod_coll_op
 
// =-=-=-=-=-=-=-
// commit the transaction
irods::error db_reg_zone_op(
    irods::plugin_context& _ctx,
    const char*            _zone_name,
    const char*            _zone_type,
    const char*            _zone_conn_info,
    const char*            _zone_comment ) {
    // =-=-=-=-=-=-=-
    // check the context
    irods::error ret = _ctx.valid();
    if ( !ret.ok() ) {
        return PASS( ret );
    }
 
    // =-=-=-=-=-=-=-
    // check the params
    if (
        !_zone_name ||
        !_zone_type ||
        !_zone_conn_info ||
        !_zone_comment ) {
        return ERROR(
                   CAT_INVALID_ARGUMENT,
                   "null parameter" );
    }
 
    // =-=-=-=-=-=-=-
    // get a postgres object from the context
    /*irods::postgres_object_ptr pg;
    ret = make_db_ptr( _ctx.fco(), pg );
    if ( !ret.ok() ) {
        return PASS( ret );
 
    }*/
 
    // =-=-=-=-=-=-=-
    // extract the icss property
//        icatSessionStruct icss;
//        _ctx.prop_map().get< icatSessionStruct >( ICSS_PROP, icss );
    char nextStr[MAX_NAME_LEN];
    char tSQL[MAX_SQL_SIZE];
    int status;
    char myTime[50];
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegZone" );
    }
 
    if ( !icss.status ) {
        return ERROR( CATALOG_NOT_CONNECTED, "catalog not connected" );
    }
 
    if ( _ctx.comm()->clientUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege" );
    }
    if ( _ctx.comm()->proxyUser.authInfo.authFlag < LOCAL_PRIV_USER_AUTH ) {
        return ERROR( CAT_INSUFFICIENT_PRIVILEGE_LEVEL, "insufficient privilege" );
    }
 
    if ( strncmp( _zone_type, "remote", 6 ) != 0 ) {
        addRErrorMsg( &_ctx.comm()->rError, 0,
                      "Currently, only zones of type 'remote' are allowed" );
        return ERROR( CAT_INVALID_ARGUMENT, "Currently, only zones of type 'remote' are allowed" );
    }
 
    // =-=-=-=-=-=-=-
    // validate the zone name does not include improper characters
    ret = validate_zone_name( _zone_name );
    if ( !ret.ok() ) {
        irods::log( ret );
        return PASS( ret );
    }
 
    /* String to get next sequence item for objects */
    cllNextValueString( "R_ObjectID", nextStr, MAX_NAME_LEN );
 
    getNowStr( myTime );
 
    if ( logSQL != 0 ) {
        rodsLog( LOG_SQL, "chlRegZone SQL 1 " );
    }
    cllBindVars[cllBindVarCount++] = _zone_name;
    cllBindVars[cllBindVarCount++] = _zone_conn_info;
    cllBindVars[cllBindVarCount++] = _zone_comment;
    cllBindVars[cllBindVarCount++] = myTime;
    cllBindVars[cllBindVarCount++] = myTime;
 
    snprintf( tSQL, MAX_SQL_SIZE,
              "insert into R_ZONE_MAIN (zone_id, zone_name, zone_type_name, zone_conn_string, r_comment, create_ts, modify_ts) values (%s, ?, 'remote', ?, ?, ?, ?)",
              nextStr );
    status =  cmlExecuteNoAnswerSql(