[working/Evergreen.git] / Open-ILS / src / c-apps / oils_sql.c

/**
        @file oils_sql.c
        @brief Utility routines for translating JSON into SQL.
*/

#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <ctype.h>
#include <dbi/dbi.h>
#include "opensrf/utils.h"
#include "opensrf/log.h"
#include "opensrf/osrf_application.h"
#include "openils/oils_utils.h"
#include "openils/oils_sql.h"

// The next four macros are OR'd together as needed to form a set
// of bitflags.  SUBCOMBO enables an extra pair of parentheses when
// nesting one UNION, INTERSECT or EXCEPT inside another.
// SUBSELECT tells us we're in a subquery, so don't add the
// terminal semicolon yet.
#define SUBCOMBO    8
#define SUBSELECT   4
#define DISABLE_I18N    2
#define SELECT_DISTINCT 1

#define AND_OP_JOIN     0
#define OR_OP_JOIN      1

struct ClassInfoStruct;
typedef struct ClassInfoStruct ClassInfo;

#define ALIAS_STORE_SIZE 16
#define CLASS_NAME_STORE_SIZE 16

struct ClassInfoStruct {
        char* alias;
        char* class_name;
        char* source_def;
        osrfHash* class_def;      // Points into IDL
        osrfHash* fields;         // Points into IDL
        osrfHash* links;          // Points into IDL

        // The remaining members are private and internal.  Client code should not
        // access them directly.

        ClassInfo* next;          // Supports linked list of joined classes
        int in_use;               // boolean

        // We usually store the alias and class name in the following arrays, and
        // point the corresponding pointers at them.  When the string is too big
        // for the array (which will probably never happen in practice), we strdup it.

        char alias_store[ ALIAS_STORE_SIZE + 1 ];
        char class_name_store[ CLASS_NAME_STORE_SIZE + 1 ];
};

struct QueryFrameStruct;
typedef struct QueryFrameStruct QueryFrame;

struct QueryFrameStruct {
        ClassInfo core;
        ClassInfo* join_list;  // linked list of classes joined to the core class
        QueryFrame* next;      // implements stack as linked list
        int in_use;            // boolean
};

static int timeout_needs_resetting;
static time_t time_next_reset;

static int verifyObjectClass ( osrfMethodContext*, const jsonObject* );

static void setXactId( osrfMethodContext* ctx );
static inline const char* getXactId( osrfMethodContext* ctx );
static inline void clearXactId( osrfMethodContext* ctx );

static jsonObject* doFieldmapperSearch ( osrfMethodContext* ctx, osrfHash* class_meta,
                jsonObject* where_hash, jsonObject* query_hash, int* err );
static jsonObject* oilsMakeFieldmapperFromResult( dbi_result, osrfHash* );
static jsonObject* oilsMakeJSONFromResult( dbi_result );

static char* searchSimplePredicate ( const char* op, const char* class_alias,
                                osrfHash* field, const jsonObject* node );
static char* searchFunctionPredicate ( const char*, osrfHash*, const jsonObject*, const char* );
static char* searchFieldTransform ( const char*, osrfHash*, const jsonObject* );
static char* searchFieldTransformPredicate ( const ClassInfo*, osrfHash*, const jsonObject*,
                const char* );
static char* searchBETWEENPredicate ( const char*, osrfHash*, const jsonObject* );
static char* searchINPredicate ( const char*, osrfHash*,
                                                                 jsonObject*, const char*, osrfMethodContext* );
static char* searchPredicate ( const ClassInfo*, osrfHash*, jsonObject*, osrfMethodContext* );
static char* searchJOIN ( const jsonObject*, const ClassInfo* left_info );
static char* searchWHERE ( const jsonObject*, const ClassInfo*, int, osrfMethodContext* );
static char* buildSELECT ( jsonObject*, jsonObject*, osrfHash*, osrfMethodContext* );
char* buildQuery( osrfMethodContext* ctx, jsonObject* query, int flags );

char* SELECT ( osrfMethodContext*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, int );

void userDataFree( void* );
static void sessionDataFree( char*, void* );
static int obj_is_true( const jsonObject* obj );
static const char* json_type( int code );
static const char* get_primitive( osrfHash* field );
static const char* get_datatype( osrfHash* field );
static int is_identifier( const char* s );
static int is_good_operator( const char* op );
static void pop_query_frame( void );
static void push_query_frame( void );
static int add_query_core( const char* alias, const char* class_name );
static inline ClassInfo* search_alias( const char* target );
static ClassInfo* search_all_alias( const char* target );
static ClassInfo* add_joined_class( const char* alias, const char* classname );
static void clear_query_stack( void );

static const jsonObject* verifyUserPCRUD( osrfMethodContext* );
static int verifyObjectPCRUD( osrfMethodContext*, const jsonObject* );
static const char* org_tree_root( osrfMethodContext* ctx );
static jsonObject* single_hash( const char* key, const char* value );

static int child_initialized = 0;   /* boolean */

static dbi_conn writehandle; /* our MASTER db connection */
static dbi_conn dbhandle; /* our CURRENT db connection */
//static osrfHash * readHandles;

// The following points to the top of a stack of QueryFrames.  It's a little
// confusing because the top level of the query is at the bottom of the stack.
static QueryFrame* curr_query = NULL;

static dbi_conn writehandle; /* our MASTER db connection */
static dbi_conn dbhandle; /* our CURRENT db connection */
//static osrfHash * readHandles;

static int max_flesh_depth = 100;

static int enforce_pcrud = 0;     // Boolean
static char* modulename = NULL;

/**
        @brief Connect to the database.
        @return A database connection if successful, or NULL if not.
*/
dbi_conn oilsConnectDB( const char* mod_name ) {

        osrfLogDebug( OSRF_LOG_MARK, "Attempting to initialize libdbi..." );
        if( dbi_initialize( NULL ) == -1 ) {
                osrfLogError( OSRF_LOG_MARK, "Unable to initialize libdbi" );
                return NULL;
        } else
                osrfLogDebug( OSRF_LOG_MARK, "... libdbi initialized." );

        char* driver = osrf_settings_host_value( "/apps/%s/app_settings/driver", mod_name );
        char* user   = osrf_settings_host_value( "/apps/%s/app_settings/database/user", mod_name );
        char* host   = osrf_settings_host_value( "/apps/%s/app_settings/database/host", mod_name );
        char* port   = osrf_settings_host_value( "/apps/%s/app_settings/database/port", mod_name );
        char* db     = osrf_settings_host_value( "/apps/%s/app_settings/database/db", mod_name );
        char* pw     = osrf_settings_host_value( "/apps/%s/app_settings/database/pw", mod_name );

        osrfLogDebug( OSRF_LOG_MARK, "Attempting to load the database driver [%s]...", driver );
        dbi_conn handle = dbi_conn_new( driver );

        if( !handle ) {
                osrfLogError( OSRF_LOG_MARK, "Error loading database driver [%s]", driver );
                return NULL;
        }
        osrfLogDebug( OSRF_LOG_MARK, "Database driver [%s] seems OK", driver );

        osrfLogInfo(OSRF_LOG_MARK, "%s connecting to database.  host=%s, "
                "port=%s, user=%s, db=%s", mod_name, host, port, user, db );

        if( host ) dbi_conn_set_option( handle, "host", host );
        if( port ) dbi_conn_set_option_numeric( handle, "port", atoi( port ));
        if( user ) dbi_conn_set_option( handle, "username", user );
        if( pw )   dbi_conn_set_option( handle, "password", pw );
        if( db )   dbi_conn_set_option( handle, "dbname", db );

        free( user );
        free( host );
        free( port );
        free( db );
        free( pw );

        if( dbi_conn_connect( handle ) < 0 ) {
                sleep( 1 );
                if( dbi_conn_connect( handle ) < 0 ) {
                        const char* errmsg;
                        dbi_conn_error( handle, &errmsg );
                        osrfLogError( OSRF_LOG_MARK, "Error connecting to database: %s", errmsg );
                        return NULL;
                }
        }

        osrfLogInfo( OSRF_LOG_MARK, "%s successfully connected to the database", mod_name );

        return handle;
}

/**
        @brief Select some options.
        @param module_name: Name of the server.
        @param do_pcrud: Boolean.  True if we are to enforce PCRUD permissions.

        This source file is used (at this writing) to implement three different servers:
        - open-ils.reporter-store
        - open-ils.pcrud
        - open-ils.cstore

        These servers behave mostly the same, but they implement different combinations of
        methods, and open-ils.pcrud enforces a permissions scheme that the other two don't.

        Here we use the server name in messages to identify which kind of server issued them.
        We use do_crud as a boolean to control whether or not to enforce the permissions scheme.
*/
void oilsSetSQLOptions( const char* module_name, int do_pcrud, int flesh_depth ) {
        if( !module_name )
                module_name = "open-ils.cstore";   // bulletproofing with a default

        if( modulename )
                free( modulename );

        modulename = strdup( module_name );
        enforce_pcrud = do_pcrud;
        max_flesh_depth = flesh_depth;
}

/**
        @brief Install a database connection.
        @param conn Pointer to a database connection.

        In some contexts, @a conn may merely provide a driver so that we can process strings
        properly, without providing an open database connection.
*/
void oilsSetDBConnection( dbi_conn conn ) {
        dbhandle = writehandle = conn;
}

/**
        @brief Get a table name, view name, or subquery for use in a FROM clause.
        @param class Pointer to the IDL class entry.
        @return A table name, a view name, or a subquery in parentheses.

        In some cases the IDL defines a class, not with a table name or a view name, but with
        a SELECT statement, which may be used as a subquery.
*/
char* oilsGetRelation( osrfHash* classdef ) {

        char* source_def = NULL;
        const char* tabledef = osrfHashGet( classdef, "tablename" );

        if( tabledef ) {
                source_def = strdup( tabledef );   // Return the name of a table or view
        } else {
                tabledef = osrfHashGet( classdef, "source_definition" );
                if( tabledef ) {
                        // Return a subquery, enclosed in parentheses
                        source_def = safe_malloc( strlen( tabledef ) + 3 );
                        source_def[ 0 ] = '(';
                        strcpy( source_def + 1, tabledef );
                        strcat( source_def, ")" );
                } else {
                        // Not found: return an error
                        const char* classname = osrfHashGet( classdef, "classname" );
                        if( !classname )
                                classname = "???";
                        osrfLogError(
                                OSRF_LOG_MARK,
                                "%s ERROR No tablename or source_definition for class \"%s\"",
                                modulename,
                                classname
                        );
                }
        }

        return source_def;
}

/**
        @brief Add datatypes from the database to the fields in the IDL.
        @param handle Handle for a database connection
        @return Zero if successful, or 1 upon error.

        For each relevant class in the IDL: ask the database for the datatype of every field.
        In particular, determine which fields are text fields and which fields are numeric
        fields, so that we know whether to enclose their values in quotes.
*/
int oilsExtendIDL( dbi_conn handle ) {
        osrfHashIterator* class_itr = osrfNewHashIterator( oilsIDL() );
        osrfHash* class = NULL;
        growing_buffer* query_buf = buffer_init( 64 );
        int results_found = 0;   // boolean

        // For each class in the IDL...
        while( (class = osrfHashIteratorNext( class_itr ) ) ) {
                const char* classname = osrfHashIteratorKey( class_itr );
                osrfHash* fields = osrfHashGet( class, "fields" );

                // If the class is virtual, ignore it
                if( str_is_true( osrfHashGet(class, "virtual") ) ) {
                        osrfLogDebug(OSRF_LOG_MARK, "Class %s is virtual, skipping", classname );
                        continue;
                }

                char* tabledef = oilsGetRelation( class );
                if( !tabledef )
                        continue;   // No such relation -- a query of it would be doomed to failure

                buffer_reset( query_buf );
                buffer_fadd( query_buf, "SELECT * FROM %s AS x WHERE 1=0;", tabledef );

                free(tabledef );

                osrfLogDebug( OSRF_LOG_MARK, "%s Investigatory SQL = %s",
                                modulename, OSRF_BUFFER_C_STR( query_buf ) );

                dbi_result result = dbi_conn_query( handle, OSRF_BUFFER_C_STR( query_buf ) );
                if( result ) {

                        results_found = 1;
                        int columnIndex = 1;
                        const char* columnName;
                        while( (columnName = dbi_result_get_field_name(result, columnIndex)) ) {

                                osrfLogInternal( OSRF_LOG_MARK, "Looking for column named [%s]...",
                                                columnName );

                                /* fetch the fieldmapper index */
                                osrfHash* _f = osrfHashGet(fields, columnName);
                                if( _f ) {

                                        osrfLogDebug(OSRF_LOG_MARK, "Found [%s] in IDL hash...", columnName);

                                        /* determine the field type and storage attributes */

                                        switch( dbi_result_get_field_type_idx( result, columnIndex )) {

                                                case DBI_TYPE_INTEGER : {

                                                        if( !osrfHashGet(_f, "primitive") )
                                                                osrfHashSet(_f, "number", "primitive");

                                                        int attr = dbi_result_get_field_attribs_idx( result, columnIndex );
                                                        if( attr & DBI_INTEGER_SIZE8 )
                                                                osrfHashSet( _f, "INT8", "datatype" );
                                                        else
                                                                osrfHashSet( _f, "INT", "datatype" );
                                                        break;
                                                }
                                                case DBI_TYPE_DECIMAL :
                                                        if( !osrfHashGet( _f, "primitive" ))
                                                                osrfHashSet( _f, "number", "primitive" );

                                                        osrfHashSet( _f, "NUMERIC", "datatype" );
                                                        break;

                                                case DBI_TYPE_STRING :
                                                        if( !osrfHashGet( _f, "primitive" ))
                                                                osrfHashSet( _f, "string", "primitive" );

                                                        osrfHashSet( _f,"TEXT", "datatype" );
                                                        break;

                                                case DBI_TYPE_DATETIME :
                                                        if( !osrfHashGet( _f, "primitive" ))
                                                                osrfHashSet( _f, "string", "primitive" );

                                                        osrfHashSet( _f, "TIMESTAMP", "datatype" );
                                                        break;

                                                case DBI_TYPE_BINARY :
                                                        if( !osrfHashGet( _f, "primitive" ))
                                                                osrfHashSet( _f, "string", "primitive" );

                                                        osrfHashSet( _f, "BYTEA", "datatype" );
                                        }

                                        osrfLogDebug(
                                                OSRF_LOG_MARK,
                                                "Setting [%s] to primitive [%s] and datatype [%s]...",
                                                columnName,
                                                osrfHashGet( _f, "primitive" ),
                                                osrfHashGet( _f, "datatype" )
                                        );
                                }
                                ++columnIndex;
                        } // end while loop for traversing columns of result
                        dbi_result_free( result  );
                } else {
                        osrfLogDebug( OSRF_LOG_MARK, "No data found for class [%s]...", classname );
                }
        } // end for each class in IDL

        buffer_free( query_buf );
        osrfHashIteratorFree( class_itr );
        child_initialized = 1;

        if( !results_found ) {
                osrfLogError( OSRF_LOG_MARK,
                        "No results found for any class -- bad database connection?" );
                return 1;
        }
        else
                return 0;
}

/**
        @brief Free an osrfHash that stores a transaction ID.
        @param blob A pointer to the osrfHash to be freed, cast to a void pointer.

        This function is a callback, to be called by the application session when it ends.
        The application session stores the osrfHash via an opaque pointer.

        If the osrfHash contains an entry for the key "xact_id", it means that an
        uncommitted transaction is pending.  Roll it back.
*/
void userDataFree( void* blob ) {
        osrfHash* hash = (osrfHash*) blob;
        if( osrfHashGet( hash, "xact_id" ) && writehandle )
                dbi_conn_query( writehandle, "ROLLBACK;" );

        osrfHashFree( hash );
}

/**
        @name Managing session data
        @brief Maintain data stored via the userData pointer of the application session.

        Currently, session-level data is stored in an osrfHash.  Other arrangements are
        possible, and some would be more efficient.  The application session calls a
        callback function to free userData before terminating.

        Currently, the only data we store at the session level is the transaction id.  By this
        means we can ensure that any pending transactions are rolled back before the application
        session terminates.
*/
/*@{*/

/**
        @brief Free an item in the application session's userData.
        @param key The name of a key for an osrfHash.
        @param item An opaque pointer to the item associated with the key.

        We store an osrfHash as userData with the application session, and arrange (by
        installing userDataFree() as a different callback) for the session to free that
        osrfHash before terminating.

        This function is a callback for freeing items in the osrfHash.  Currently we store
        two things:
        - Transaction id of a pending transaction; a character string.  Key: "xact_id".
        - Authkey; a character string.  Key: "authkey".
        - User object from the authentication server; a jsonObject.  Key: "user_login".

        If we ever store anything else in userData, we will need to revisit this function so
        that it will free whatever else needs freeing.
*/
static void sessionDataFree( char* key, void* item ) {
        if( !strcmp( key, "xact_id" )
             || !strcmp( key, "authkey" ) ) {
                free( item );
        } else if( !strcmp( key, "user_login" ) )
                jsonObjectFree( (jsonObject*) item );
}

/**
        @brief Save a transaction id.
        @param ctx Pointer to the method context.

        Save the session_id of the current application session as a transaction id.
*/
static void setXactId( osrfMethodContext* ctx ) {
        if( ctx && ctx->session ) {
                osrfAppSession* session = ctx->session;

                osrfHash* cache = session->userData;

                // If the session doesn't already have a hash, create one.  Make sure
                // that the application session frees the hash when it terminates.
                if( NULL == cache ) {
                        session->userData = cache = osrfNewHash();
                        osrfHashSetCallback( cache, &sessionDataFree );
                        ctx->session->userDataFree = &userDataFree;
                }

                // Save the transaction id in the hash, with the key "xact_id"
                osrfHashSet( cache, strdup( session->session_id ), "xact_id" );
        }
}

/**
        @brief Get the transaction ID for the current transaction, if any.
        @param ctx Pointer to the method context.
        @return Pointer to the transaction ID.

        The return value points to an internal buffer, and will become invalid upon issuing
        a commit or rollback.
*/
static inline const char* getXactId( osrfMethodContext* ctx ) {
        if( ctx && ctx->session && ctx->session->userData )
                return osrfHashGet( (osrfHash*) ctx->session->userData, "xact_id" );
        else
                return NULL;
}

/**
        @brief Clear the current transaction id.
        @param ctx Pointer to the method context.
*/
static inline void clearXactId( osrfMethodContext* ctx ) {
        if( ctx && ctx->session && ctx->session->userData )
                osrfHashRemove( ctx->session->userData, "xact_id" );
}
/*@}*/

/**
        @brief Save the user's login in the userData for the current application session.
        @param ctx Pointer to the method context.
        @param user_login Pointer to the user login object to be cached (we cache the original,
        not a copy of it).

        If @a user_login is NULL, remove the user login if one is already cached.
*/
static void setUserLogin( osrfMethodContext* ctx, jsonObject* user_login ) {
        if( ctx && ctx->session ) {
                osrfAppSession* session = ctx->session;

                osrfHash* cache = session->userData;

                // If the session doesn't already have a hash, create one.  Make sure
                // that the application session frees the hash when it terminates.
                if( NULL == cache ) {
                        session->userData = cache = osrfNewHash();
                        osrfHashSetCallback( cache, &sessionDataFree );
                        ctx->session->userDataFree = &userDataFree;
                }

                if( user_login )
                        osrfHashSet( cache, user_login, "user_login" );
                else
                        osrfHashRemove( cache, "user_login" );
        }
}

/**
        @brief Get the user login object for the current application session, if any.
        @param ctx Pointer to the method context.
        @return Pointer to the user login object if found; otherwise NULL.

        The user login object was returned from the authentication server, and then cached so
        we don't have to call the authentication server again for the same user.
*/
static const jsonObject* getUserLogin( osrfMethodContext* ctx ) {
        if( ctx && ctx->session && ctx->session->userData )
                return osrfHashGet( (osrfHash*) ctx->session->userData, "user_login" );
        else
                return NULL;
}

/**
        @brief Save a copy of an authkey in the userData of the current application session.
        @param ctx Pointer to the method context.
        @param authkey The authkey to be saved.

        If @a authkey is NULL, remove the authkey if one is already cached.
*/
static void setAuthkey( osrfMethodContext* ctx, const char* authkey ) {
        if( ctx && ctx->session && authkey ) {
                osrfAppSession* session = ctx->session;
                osrfHash* cache = session->userData;

                // If the session doesn't already have a hash, create one.  Make sure
                // that the application session frees the hash when it terminates.
                if( NULL == cache ) {
                        session->userData = cache = osrfNewHash();
                        osrfHashSetCallback( cache, &sessionDataFree );
                        ctx->session->userDataFree = &userDataFree;
                }

                // Save the transaction id in the hash, with the key "xact_id"
                if( authkey && *authkey )
                        osrfHashSet( cache, strdup( authkey ), "authkey" );
                else
                        osrfHashRemove( cache, "authkey" );
        }
}

/**
        @brief Reset the login timeout.
        @param authkey The authentication key for the current login session.
        @param now The current time.
        @return Zero if successful, or 1 if not.

        Tell the authentication server to reset the timeout so that the login session won't
        expire for a while longer.

        We could dispense with the @a now parameter by calling time().  But we just called
        time() in order to decide whether to reset the timeout, so we might as well reuse
        the result instead of calling time() again.
*/
static int reset_timeout( const char* authkey, time_t now ) {
        jsonObject* auth_object = jsonNewObject( authkey );

        // Ask the authentication server to reset the timeout.  It returns an event
        // indicating success or failure.
        jsonObject* result = oilsUtilsQuickReq( "open-ils.auth",
                "open-ils.auth.session.reset_timeout", auth_object );
        jsonObjectFree( auth_object );

        if( !result || result->type != JSON_HASH ) {
                osrfLogError( OSRF_LOG_MARK,
                         "Unexpected object type receieved from open-ils.auth.session.reset_timeout" );
                jsonObjectFree( result );
                return 1;       // Not the right sort of object returned
        }

        const jsonObject* ilsevent = jsonObjectGetKey( result, "ilsevent" );
        if( !ilsevent || ilsevent->type != JSON_NUMBER ) {
                osrfLogError( OSRF_LOG_MARK, "ilsevent is absent or malformed" );
                jsonObjectFree( result );
                return 1;    // Return code from method not available
        }

        if( jsonObjectGetNumber( ilsevent ) != 0.0 ) {
                const char* desc = jsonObjectGetString( jsonObjectGetKey( result, "desc" ) );
                if( !desc )
                        desc = "(No reason available)";    // failsafe; shouldn't happen
                osrfLogInfo( OSRF_LOG_MARK, "Failure to reset timeout: %s", desc );
                jsonObjectFree( result );
                return 1;
        }

        // Revise our local proxy for the timeout deadline
        // by a smallish fraction of the timeout interval
        const char* timeout = jsonObjectGetString( jsonObjectGetKey( result, "payload" ) );
        if( !timeout )
                timeout = "1";   // failsafe; shouldn't happen
        time_next_reset = now + atoi( timeout ) / 15;

        jsonObjectFree( result );
        return 0;     // Successfully reset timeout
}

/**
        @brief Get the authkey string for the current application session, if any.
        @param ctx Pointer to the method context.
        @return Pointer to the cached authkey if found; otherwise NULL.

        If present, the authkey string was cached from a previous method call.
*/
static const char* getAuthkey( osrfMethodContext* ctx ) {
        if( ctx && ctx->session && ctx->session->userData ) {
                const char* authkey = osrfHashGet( (osrfHash*) ctx->session->userData, "authkey" );

                // Possibly reset the authentication timeout to keep the login alive.  We do so
                // no more than once per method call, and not at all if it has been only a short
                // time since the last reset.

                // Here we reset explicitly, if at all.  We also implicitly reset the timeout
                // whenever we call the "open-ils.auth.session.retrieve" method.
                if( timeout_needs_resetting ) {
                        time_t now = time( NULL );
                        if( now >= time_next_reset && reset_timeout( authkey, now ) )
                                authkey = NULL;    // timeout has apparently expired already
                }

                timeout_needs_resetting = 0;
                return authkey;
        }
        else
                return NULL;
}

/**
        @brief Implement the transaction.begin method.
        @param ctx Pointer to the method context.
        @return Zero if successful, or -1 upon error.

        Start a transaction.  Save a transaction ID for future reference.

        Method parameters:
        - authkey (PCRUD only)

        Return to client: Transaction ID
*/
int beginTransaction( osrfMethodContext* ctx ) {
        if(osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
                return -1;
        }

        if( enforce_pcrud ) {
                timeout_needs_resetting = 1;
                const jsonObject* user = verifyUserPCRUD( ctx );
                if( !user )
                        return -1;
        }

        dbi_result result = dbi_conn_query( writehandle, "START TRANSACTION;" );
        if( !result ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Error starting transaction", modulename );
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
                                "osrfMethodException", ctx->request, "Error starting transaction" );
                return -1;
        } else {
                setXactId( ctx );
                jsonObject* ret = jsonNewObject( getXactId( ctx ) );
                osrfAppRespondComplete( ctx, ret );
                jsonObjectFree( ret );
        }
        return 0;
}

/**
        @brief Implement the savepoint.set method.
        @param ctx Pointer to the method context.
        @return Zero if successful, or -1 if not.

        Issue a SAVEPOINT to the database server.

        Method parameters:
        - authkey (PCRUD only)
        - savepoint name

        Return to client: Savepoint name
*/
int setSavepoint( osrfMethodContext* ctx ) {
        if(osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
                return -1;
        }

        int spNamePos = 0;
        if( enforce_pcrud ) {
                spNamePos = 1;
                timeout_needs_resetting = 1;
                const jsonObject* user = verifyUserPCRUD( ctx );
                if( !user )
                        return -1;
        }

        // Verify that a transaction is pending
        const char* trans_id = getXactId( ctx );
        if( NULL == trans_id ) {
                osrfAppSessionStatus(
                        ctx->session,
                        OSRF_STATUS_INTERNALSERVERERROR,
                        "osrfMethodException",
                        ctx->request,
                        "No active transaction -- required for savepoints"
                );
                return -1;
        }

        // Get the savepoint name from the method params
        const char* spName = jsonObjectGetString( jsonObjectGetIndex(ctx->params, spNamePos) );

        dbi_result result = dbi_conn_queryf( writehandle, "SAVEPOINT \"%s\";", spName );
        if( !result ) {
                osrfLogError(
                        OSRF_LOG_MARK,
                        "%s: Error creating savepoint %s in transaction %s",
                        modulename,
                        spName,
                        trans_id
                );
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
                                "osrfMethodException", ctx->request, "Error creating savepoint" );
                return -1;
        } else {
                jsonObject* ret = jsonNewObject( spName );
                osrfAppRespondComplete( ctx, ret );
                jsonObjectFree( ret  );
        }
        return 0;
}

/**
        @brief Implement the savepoint.release method.
        @param ctx Pointer to the method context.
        @return Zero if successful, or -1 if not.

        Issue a RELEASE SAVEPOINT to the database server.

        Method parameters:
        - authkey (PCRUD only)
        - savepoint name

        Return to client: Savepoint name
*/
int releaseSavepoint( osrfMethodContext* ctx ) {
        if(osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
                return -1;
        }

        int spNamePos = 0;
        if( enforce_pcrud ) {
                spNamePos = 1;
                timeout_needs_resetting = 1;
                const jsonObject* user = verifyUserPCRUD( ctx );
                if(  !user )
                        return -1;
        }

        // Verify that a transaction is pending
        const char* trans_id = getXactId( ctx );
        if( NULL == trans_id ) {
                osrfAppSessionStatus(
                        ctx->session,
                        OSRF_STATUS_INTERNALSERVERERROR,
                        "osrfMethodException",
                        ctx->request,
                        "No active transaction -- required for savepoints"
                );
                return -1;
        }

        // Get the savepoint name from the method params
        const char* spName = jsonObjectGetString( jsonObjectGetIndex(ctx->params, spNamePos) );

        dbi_result result = dbi_conn_queryf( writehandle, "RELEASE SAVEPOINT \"%s\";", spName );
        if( !result ) {
                osrfLogError(
                        OSRF_LOG_MARK,
                        "%s: Error releasing savepoint %s in transaction %s",
                        modulename,
                        spName,
                        trans_id
                );
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
                                "osrfMethodException", ctx->request, "Error releasing savepoint" );
                return -1;
        } else {
                jsonObject* ret = jsonNewObject( spName );
                osrfAppRespondComplete( ctx, ret );
                jsonObjectFree( ret );
        }
        return 0;
}

/**
        @brief Implement the savepoint.rollback method.
        @param ctx Pointer to the method context.
        @return Zero if successful, or -1 if not.

        Issue a ROLLBACK TO SAVEPOINT to the database server.

        Method parameters:
        - authkey (PCRUD only)
        - savepoint name

        Return to client: Savepoint name
*/
int rollbackSavepoint( osrfMethodContext* ctx ) {
        if(osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
                return -1;
        }

        int spNamePos = 0;
        if( enforce_pcrud ) {
                spNamePos = 1;
                timeout_needs_resetting = 1;
                const jsonObject* user = verifyUserPCRUD( ctx );
                if( !user )
                        return -1;
        }

        // Verify that a transaction is pending
        const char* trans_id = getXactId( ctx );
        if( NULL == trans_id ) {
                osrfAppSessionStatus(
                        ctx->session,
                        OSRF_STATUS_INTERNALSERVERERROR,
                        "osrfMethodException",
                        ctx->request,
                        "No active transaction -- required for savepoints"
                );
                return -1;
        }

        // Get the savepoint name from the method params
        const char* spName = jsonObjectGetString( jsonObjectGetIndex(ctx->params, spNamePos) );

        dbi_result result = dbi_conn_queryf( writehandle, "ROLLBACK TO SAVEPOINT \"%s\";", spName );
        if( !result ) {
                osrfLogError(
                        OSRF_LOG_MARK,
                        "%s: Error rolling back savepoint %s in transaction %s",
                        modulename,
                        spName,
                        trans_id
                );
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
                                "osrfMethodException", ctx->request, "Error rolling back savepoint" );
                return -1;
        } else {
                jsonObject* ret = jsonNewObject( spName );
                osrfAppRespondComplete( ctx, ret );
                jsonObjectFree( ret );
        }
        return 0;
}

/**
        @brief Implement the transaction.commit method.
        @param ctx Pointer to the method context.
        @return Zero if successful, or -1 if not.

        Issue a COMMIT to the database server.

        Method parameters:
        - authkey (PCRUD only)

        Return to client: Transaction ID.
*/
int commitTransaction( osrfMethodContext* ctx ) {
        if(osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
                return -1;
        }

        if( enforce_pcrud ) {
                timeout_needs_resetting = 1;
                const jsonObject* user = verifyUserPCRUD( ctx );
                if( !user )
                        return -1;
        }

        // Verify that a transaction is pending
        const char* trans_id = getXactId( ctx );
        if( NULL == trans_id ) {
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
                                "osrfMethodException", ctx->request, "No active transaction to commit" );
                return -1;
        }

        dbi_result result = dbi_conn_query( writehandle, "COMMIT;" );
        if( !result ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Error committing transaction", modulename );
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
                                "osrfMethodException", ctx->request, "Error committing transaction" );
                return -1;
        } else {
                jsonObject* ret = jsonNewObject( trans_id );
                osrfAppRespondComplete( ctx, ret );
                jsonObjectFree( ret );
                clearXactId( ctx );
        }
        return 0;
}

/**
        @brief Implement the transaction.rollback method.
        @param ctx Pointer to the method context.
        @return Zero if successful, or -1 if not.

        Issue a ROLLBACK to the database server.

        Method parameters:
        - authkey (PCRUD only)

        Return to client: Transaction ID
*/
int rollbackTransaction( osrfMethodContext* ctx ) {
        if( osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
                return -1;
        }

        if( enforce_pcrud ) {
                timeout_needs_resetting = 1;
                const jsonObject* user = verifyUserPCRUD( ctx );
                if( !user )
                        return -1;
        }

        // Verify that a transaction is pending
        const char* trans_id = getXactId( ctx );
        if( NULL == trans_id ) {
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
                                "osrfMethodException", ctx->request, "No active transaction to roll back" );
                return -1;
        }

        dbi_result result = dbi_conn_query( writehandle, "ROLLBACK;" );
        if( !result ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Error rolling back transaction", modulename );
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
                                "osrfMethodException", ctx->request, "Error rolling back transaction" );
                return -1;
        } else {
                jsonObject* ret = jsonNewObject( trans_id );
                osrfAppRespondComplete( ctx, ret );
                jsonObjectFree( ret );
                clearXactId( ctx );
        }
        return 0;
}

/**
        @brief Implement the "search" method.
        @param ctx Pointer to the method context.
        @return Zero if successful, or -1 if not.

        Method parameters:
        - authkey (PCRUD only)
        - WHERE clause, as jsonObject
        - Other SQL clause(s), as a JSON_HASH: joins, SELECT list, LIMIT, etc.

        Return to client: rows of the specified class that satisfy a specified WHERE clause.
        Optionally flesh linked fields.
*/
int doSearch( osrfMethodContext* ctx ) {
        if( osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
                return -1;
        }

        if( enforce_pcrud )
                timeout_needs_resetting = 1;

        jsonObject* where_clause;
        jsonObject* rest_of_query;

        if( enforce_pcrud ) {
                where_clause  = jsonObjectGetIndex( ctx->params, 1 );
                rest_of_query = jsonObjectGetIndex( ctx->params, 2 );
        } else {
                where_clause  = jsonObjectGetIndex( ctx->params, 0 );
                rest_of_query = jsonObjectGetIndex( ctx->params, 1 );
        }

        // Get the class metadata
        osrfHash* method_meta = (osrfHash*) ctx->method->userData;
        osrfHash* class_meta = osrfHashGet( method_meta, "class" );

        // Do the query
        int err = 0;
        jsonObject* obj = doFieldmapperSearch( ctx, class_meta, where_clause, rest_of_query, &err );
        if( err ) {
                osrfAppRespondComplete( ctx, NULL );
                return -1;
        }

        // Return each row to the client (except that some may be suppressed by PCRUD)
        jsonObject* cur = 0;
        unsigned long res_idx = 0;
        while((cur = jsonObjectGetIndex( obj, res_idx++ ) )) {
                if( enforce_pcrud && !verifyObjectPCRUD( ctx, cur ))
                        continue;
                osrfAppRespond( ctx, cur );
        }
        jsonObjectFree( obj );

        osrfAppRespondComplete( ctx, NULL );
        return 0;
}

/**
        @brief Implement the "id_list" method.
        @param ctx Pointer to the method context.
        @param err Pointer through which to return an error code.
        @return Zero if successful, or -1 if not.

        Method parameters:
        - authkey (PCRUD only)
        - WHERE clause, as jsonObject
        - Other SQL clause(s), as a JSON_HASH: joins, LIMIT, etc.

        Return to client: The primary key values for all rows of the relevant class that
        satisfy a specified WHERE clause.

        This method relies on the assumption that every class has a primary key consisting of
        a single column.
*/
int doIdList( osrfMethodContext* ctx ) {
        if( osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
                return -1;
        }

        if( enforce_pcrud )
                timeout_needs_resetting = 1;

        jsonObject* where_clause;
        jsonObject* rest_of_query;

        // We use the where clause without change.  But we need to massage the rest of the
        // query, so we work with a copy of it instead of modifying the original.

        if( enforce_pcrud ) {
                where_clause  = jsonObjectGetIndex( ctx->params, 1 );
                rest_of_query = jsonObjectClone( jsonObjectGetIndex( ctx->params, 2 ) );
        } else {
                where_clause  = jsonObjectGetIndex( ctx->params, 0 );
                rest_of_query = jsonObjectClone( jsonObjectGetIndex( ctx->params, 1 ) );
        }

        // Eliminate certain SQL clauses, if present.
        if( rest_of_query ) {
                jsonObjectRemoveKey( rest_of_query, "select" );
                jsonObjectRemoveKey( rest_of_query, "no_i18n" );
                jsonObjectRemoveKey( rest_of_query, "flesh" );
                jsonObjectRemoveKey( rest_of_query, "flesh_columns" );
        } else {
                rest_of_query = jsonNewObjectType( JSON_HASH );
        }

        jsonObjectSetKey( rest_of_query, "no_i18n", jsonNewBoolObject( 1 ) );

        // Get the class metadata
        osrfHash* method_meta = (osrfHash*) ctx->method->userData;
        osrfHash* class_meta = osrfHashGet( method_meta, "class" );

        // Build a SELECT list containing just the primary key,
        // i.e. like { "classname":["keyname"] }
        jsonObject* col_list_obj = jsonNewObjectType( JSON_ARRAY );

        // Load array with name of primary key
        jsonObjectPush( col_list_obj, jsonNewObject( osrfHashGet( class_meta, "primarykey" ) ) );
        jsonObject* select_clause = jsonNewObjectType( JSON_HASH );
        jsonObjectSetKey( select_clause, osrfHashGet( class_meta, "classname" ), col_list_obj );

        jsonObjectSetKey( rest_of_query, "select", select_clause );

        // Do the query
        int err = 0;
        jsonObject* obj =
                doFieldmapperSearch( ctx, class_meta, where_clause, rest_of_query, &err );

        jsonObjectFree( rest_of_query );
        if( err ) {
                osrfAppRespondComplete( ctx, NULL );
                return -1;
        }

        // Return each primary key value to the client
        jsonObject* cur;
        unsigned long res_idx = 0;
        while((cur = jsonObjectGetIndex( obj, res_idx++ ) )) {
                if( enforce_pcrud && !verifyObjectPCRUD( ctx, cur ))
                        continue;        // Suppress due to lack of permission
                else
                        osrfAppRespond( ctx,
                                oilsFMGetObject( cur, osrfHashGet( class_meta, "primarykey" ) ) );
        }

        jsonObjectFree( obj );
        osrfAppRespondComplete( ctx, NULL );
        return 0;
}

/**
        @brief Verify that we have a valid class reference.
        @param ctx Pointer to the method context.
        @param param Pointer to the method parameters.
        @return 1 if the class reference is valid, or zero if it isn't.

        The class of the method params must match the class to which the method id devoted.
        For PCRUD there are additional restrictions.
*/
static int verifyObjectClass ( osrfMethodContext* ctx, const jsonObject* param ) {

        osrfHash* method_meta = (osrfHash*) ctx->method->userData;
        osrfHash* class = osrfHashGet( method_meta, "class" );

        // Compare the method's class to the parameters' class
        if( !param->classname || (strcmp( osrfHashGet(class, "classname"), param->classname ))) {

                // Oops -- they don't match.  Complain.
                growing_buffer* msg = buffer_init( 128 );
                buffer_fadd(
                        msg,
                        "%s: %s method for type %s was passed a %s",
                        modulename,
                        osrfHashGet( method_meta, "methodtype" ),
                        osrfHashGet( class, "classname" ),
                        param->classname ? param->classname : "(null)"
                );

                char* m = buffer_release( msg );
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_BADREQUEST, "osrfMethodException",
                                ctx->request, m );
                free( m );

                return 0;
        }

        if( enforce_pcrud )
                return verifyObjectPCRUD( ctx, param );
        else
                return 1;
}

/**
        @brief (PCRUD only) Verify that the user is properly logged in.
        @param ctx Pointer to the method context.
        @return If the user is logged in, a pointer to the user object from the authentication
        server; otherwise NULL.
*/
static const jsonObject* verifyUserPCRUD( osrfMethodContext* ctx ) {

        // Get the authkey (the first method parameter)
        const char* auth = jsonObjectGetString( jsonObjectGetIndex( ctx->params, 0 ) );

        // See if we have the same authkey, and a user object,
        // locally cached from a previous call
        const char* cached_authkey = getAuthkey( ctx );
        if( cached_authkey && !strcmp( cached_authkey, auth ) ) {
                const jsonObject* cached_user = getUserLogin( ctx );
                if( cached_user )
                        return cached_user;
        }

        // We have no matching authentication data in the cache.  Authenticate from scratch.
        jsonObject* auth_object = jsonNewObject( auth );

        // Fetch the user object from the authentication server
        jsonObject* user = oilsUtilsQuickReq( "open-ils.auth", "open-ils.auth.session.retrieve",
                        auth_object );
        jsonObjectFree( auth_object );

        if( !user->classname || strcmp(user->classname, "au" )) {

                growing_buffer* msg = buffer_init( 128 );
                buffer_fadd(
                        msg,
                        "%s: permacrud received a bad auth token: %s",
                        modulename,
                        auth
                );

                char* m = buffer_release( msg );
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_UNAUTHORIZED, "osrfMethodException",
                                ctx->request, m );

                free( m );
                jsonObjectFree( user );
                user = NULL;
        }

        setUserLogin( ctx, user );
        setAuthkey( ctx, auth );

        // Allow ourselves up to a second before we have to reset the login timeout.
        // It would be nice to use some fraction of the timeout interval enforced by the
        // authentication server, but that value is not readily available at this point.
        // Instead, we use a conservative default interval.
        time_next_reset = time( NULL ) + 1;

        return user;
}

/**
        @brief For PCRUD: Determine whether the current user may access the current row.
        @param ctx Pointer to the method context.
        @param obj Pointer to the row being potentially accessed.
        @return 1 if access is permitted, or 0 if it isn't.

        The @a obj parameter points to a JSON_HASH of column values, keyed on column name.
*/
static int verifyObjectPCRUD (  osrfMethodContext* ctx, const jsonObject* obj ) {

        dbhandle = writehandle;

        // Figure out what class and method are involved
        osrfHash* method_metadata = (osrfHash*) ctx->method->userData;
        osrfHash* class = osrfHashGet( method_metadata, "class" );
        const char* method_type = osrfHashGet( method_metadata, "methodtype" );

        // Set fetch to 1 in all cases except for inserts, meaning that for local or foreign
        // contexts we will do another lookup of the current row, even if we already have a
        // previously fetched row image, because the row image in hand may not include the
        // foreign key(s) that we need.

        // This is a quick fix with a bludgeon.  There are ways to avoid the extra lookup,
        // but they aren't implemented yet.

        int fetch = 0;
        if( *method_type == 's' || *method_type == 'i' ) {
                method_type = "retrieve"; // search and id_list are equivalent to retrieve for this
                fetch = 1;
        } else if( *method_type == 'u' || *method_type == 'd' ) {
                fetch = 1; // MUST go to the db for the object for update and delete
        }

        // Get the appropriate permacrud entry from the IDL, depending on method type
        osrfHash* pcrud = osrfHashGet( osrfHashGet( class, "permacrud" ), method_type );
        if( !pcrud ) {
                // No permacrud for this method type on this class

                growing_buffer* msg = buffer_init( 128 );
                buffer_fadd(
                        msg,
                        "%s: %s on class %s has no permacrud IDL entry",
                        modulename,
                        osrfHashGet( method_metadata, "methodtype" ),
                        osrfHashGet( class, "classname" )
                );

                char* m = buffer_release( msg );
                osrfAppSessionStatus( ctx->session, OSRF_STATUS_FORBIDDEN,
                                "osrfMethodException", ctx->request, m );

                free( m );

                return 0;
        }

        // Get the user id, and make sure the user is logged in
        const jsonObject* user = verifyUserPCRUD( ctx );
        if( !user )
                return 0;    // Not logged in?  No access.

        int userid = atoi( oilsFMGetStringConst( user, "id" ) );

        // Get a list of permissions from the permacrud entry.
        osrfStringArray* permission = osrfHashGet( pcrud, "permission" );

        // Build a list of org units that own the row.  This is fairly convoluted because there
        // are several different ways that an org unit may own the row, as defined by the
        // permacrud entry.

        // Local context means that the row includes a foreign key pointing to actor.org_unit,
        // identifying an owning org_unit..
        osrfStringArray* local_context = osrfHashGet( pcrud, "local_context" );

        // Foreign context adds a layer of indirection.  The row points to some other row that
        // an org unit may own.  The "jump" attribute, if present, adds another layer of
        // indirection.
        osrfHash* foreign_context = osrfHashGet( pcrud, "foreign_context" );

        // The following string array stores the list of org units.  (We don't have a thingie
        // for storing lists of integers, so we fake it with a list of strings.)
        osrfStringArray* context_org_array = osrfNewStringArray( 1 );

        int err = 0;
        const char* pkey_value = NULL;
        if( str_is_true( osrfHashGet(pcrud, "global_required") ) ) {
                // If the global_required attribute is present and true, then the only owning
                // org unit is the root org unit, i.e. the one with no parent.
                osrfLogDebug( OSRF_LOG_MARK,
                                "global-level permissions required, fetching top of the org tree" );

                // check for perm at top of org tree
                const char* org_tree_root_id = org_tree_root( ctx );
                if( org_tree_root_id ) {
                        osrfStringArrayAdd( context_org_array, org_tree_root_id );
                        osrfLogDebug( OSRF_LOG_MARK, "top of the org tree is %s", org_tree_root_id );
                } else  {
                        osrfStringArrayFree( context_org_array );
                        return 0;
                }

        } else {
                // If the global_required attribute is absent or false, then we look for
                // local and/or foreign context.  In order to find the relevant foreign
                // keys, we must either read the relevant row from the database, or look at
                // the image of the row that we already have in memory.

                // (Herein lies a bug.  Even if we have an image of the row in memory, that
                // image may not include the foreign key column(s) that we need.)

            osrfLogDebug( OSRF_LOG_MARK, "global-level permissions not required, "
                                "fetching context org ids" );
            const char* pkey = osrfHashGet( class, "primarykey" );
                jsonObject *param = NULL;

                if( obj->classname ) {
                        pkey_value = oilsFMGetStringConst( obj, pkey );
                        if( !fetch )
                                param = jsonObjectClone( obj );
                        osrfLogDebug( OSRF_LOG_MARK, "Object supplied, using primary key value of %s",
                                        pkey_value );
                } else {
                        pkey_value = jsonObjectGetString( obj );
                        fetch = 1;
                        osrfLogDebug( OSRF_LOG_MARK, "Object not supplied, using primary key value "
                                        "of %s and retrieving from the database", pkey_value );
                }

                if( fetch ) {
                        // Fetch the row so that we can look at the foreign key(s)
                        jsonObject* _tmp_params = single_hash( pkey, pkey_value );
                        jsonObject* _list = doFieldmapperSearch( ctx, class, _tmp_params, NULL, &err );
                        jsonObjectFree( _tmp_params );

                        param = jsonObjectExtractIndex( _list, 0 );
                        jsonObjectFree( _list );
                }

                if( !param ) {
                        // The row doesn't exist.  Complain, and deny access.
                        osrfLogDebug( OSRF_LOG_MARK,
                                        "Object not found in the database with primary key %s of %s",
                                        pkey, pkey_value );

                        growing_buffer* msg = buffer_init( 128 );
                        buffer_fadd(
                                msg,
                                "%s: no object found with primary key %s of %s",
                                modulename,
                                pkey,
                                pkey_value
                        );

                        char* m = buffer_release( msg );
                        osrfAppSessionStatus(
                                ctx->session,
                                OSRF_STATUS_INTERNALSERVERERROR,
                                "osrfMethodException",
                                ctx->request,
                                m
                        );

                        free( m );
                        return 0;
                }

                if( local_context && local_context->size > 0 ) {
                        // The IDL provides a list of column names for the foreign keys denoting
                        // local context.  Look up the value of each one, and if it isn't null,
                        // add it to the list of org units.
                        osrfLogDebug( OSRF_LOG_MARK, "%d class-local context field(s) specified",
                                        local_context->size );
                        int i = 0;
                        const char* lcontext = NULL;
                        while ( (lcontext = osrfStringArrayGetString(local_context, i++)) ) {
                                const char* fkey_value = oilsFMGetStringConst( param, lcontext );
                                if( fkey_value ) {    // if not null
                                        osrfStringArrayAdd( context_org_array, fkey_value );
                                        osrfLogDebug(
                                                OSRF_LOG_MARK,
                                                "adding class-local field %s (value: %s) to the context org list",
                                                lcontext,
                                                osrfStringArrayGetString( context_org_array, context_org_array->size - 1 )
                                        );
                                }
                        }
                }

                if( foreign_context ) {
                        unsigned long class_count = osrfHashGetCount( foreign_context );
                        osrfLogDebug( OSRF_LOG_MARK, "%d foreign context classes(s) specified", class_count );

                        if( class_count > 0 ) {

                                // The IDL provides a list of foreign key columns pointing to rows that
                                // an org unit may own.  Follow each link, identify the owning org unit,
                                // and add it to the list.
                                osrfHash* fcontext = NULL;
                                osrfHashIterator* class_itr = osrfNewHashIterator( foreign_context );
                                while( (fcontext = osrfHashIteratorNext( class_itr )) ) {
                                        // For each class to which a foreign key points:
                                        const char* class_name = osrfHashIteratorKey( class_itr );
                                        osrfHash* fcontext = osrfHashGet( foreign_context, class_name );

                                        osrfLogDebug(
                                                OSRF_LOG_MARK,
                                                "%d foreign context fields(s) specified for class %s",
                                                ((osrfStringArray*)osrfHashGet(fcontext,"context"))->size,
                                                class_name
                                        );

                                        // Get the name of the key field in the foreign table
                                        const char* foreign_pkey = osrfHashGet( fcontext, "field" );

                                        // Get the value of the foreign key pointing to the foreign table
                                        char* foreign_pkey_value =
                                                        oilsFMGetString( param, osrfHashGet( fcontext, "fkey" ));
                                        if( !foreign_pkey_value )
                                                continue;    // Foreign key value is null; skip it

                                        // Look up the row to which the foreign key points
                                        jsonObject* _tmp_params = single_hash( foreign_pkey, foreign_pkey_value );
                                        jsonObject* _list = doFieldmapperSearch(
                                                ctx, osrfHashGet( oilsIDL(), class_name ), _tmp_params, NULL, &err );

                                        jsonObject* _fparam = NULL;
                                        if( _list && JSON_ARRAY == _list->type && _list->size > 0 )
                                                _fparam = jsonObjectExtractIndex( _list, 0 );

                                        jsonObjectFree( _tmp_params );
                                        jsonObjectFree( _list );

                                        // At this point _fparam either points to the row identified by the
                                        // foreign key, or it's NULL (no such row found).

                                        osrfStringArray* jump_list = osrfHashGet( fcontext, "jump" );

                                        const char* bad_class = NULL;  // For noting failed lookups
                                        if( ! _fparam )
                                                bad_class = class_name;    // Referenced row not found
                                        else if( jump_list ) {
                                                // Follow a chain of rows, linked by foreign keys, to find an owner
                                                const char* flink = NULL;
                                                int k = 0;
                                                while ( (flink = osrfStringArrayGetString(jump_list, k++)) && _fparam ) {
                                                        // For each entry in the jump list.  Each entry (i.e. flink) is
                                                        // the name of a foreign key column in the current row.

                                                        // From the IDL, get the linkage information for the next jump
                                                        osrfHash* foreign_link_hash =
                                                                        oilsIDLFindPath( "/%s/links/%s", _fparam->classname, flink );

                                                        // Get the class metadata for the class
                                                        // to which the foreign key points
                                                        osrfHash* foreign_class_meta = osrfHashGet( oilsIDL(),
                                                                        osrfHashGet( foreign_link_hash, "class" ));

                                                        // Get the name of the referenced key of that class
                                                        foreign_pkey = osrfHashGet( foreign_link_hash, "key" );

                                                        // Get the value of the foreign key pointing to that class
                                                        free( foreign_pkey_value );
                                                        foreign_pkey_value = oilsFMGetString( _fparam, flink );
                                                        if( !foreign_pkey_value )
                                                                break;    // Foreign key is null; quit looking

                                                        // Build a WHERE clause for the lookup
                                                        _tmp_params = single_hash( foreign_pkey, foreign_pkey_value );

                                                        // Do the lookup
                                                        _list = doFieldmapperSearch( ctx, foreign_class_meta,
                                                                        _tmp_params, NULL, &err );

                                                        // Get the resulting row
                                                        jsonObjectFree( _fparam );
                                                        if( _list && JSON_ARRAY == _list->type && _list->size > 0 )
                                                                _fparam = jsonObjectExtractIndex( _list, 0 );
                                                        else {
                                                                // Referenced row not found
                                                                _fparam = NULL;
                                                                bad_class = osrfHashGet( foreign_link_hash, "class" );
                                                        }

                                                        jsonObjectFree( _tmp_params );
                                                        jsonObjectFree( _list );
                                                }
                                        }

                                        if( bad_class ) {

                                                // We had a foreign key pointing to such-and-such a row, but then
                                                // we couldn't fetch that row.  The data in the database are in an
                                                // inconsistent state; the database itself may even be corrupted.
                                                growing_buffer* msg = buffer_init( 128 );
                                                buffer_fadd(
                                                        msg,
                                                        "%s: no object of class %s found with primary key %s of %s",
                                                        modulename,
                                                        bad_class,
                                                        foreign_pkey,
                                                        foreign_pkey_value ? foreign_pkey_value : "(null)"
                                                );

                                                char* m = buffer_release( msg );
                                                osrfAppSessionStatus(
                                                        ctx->session,
                                                        OSRF_STATUS_INTERNALSERVERERROR,
                                                        "osrfMethodException",
                                                        ctx->request,
                                                        m
                                                );

                                                free( m );
                                                osrfHashIteratorFree( class_itr );
                                                free( foreign_pkey_value );
                                                jsonObjectFree( param );

                                                return 0;
                                        }

                                        free( foreign_pkey_value );

                                        if( _fparam ) {
                                                // Examine each context column of the foreign row,
                                                // and add its value to the list of org units.
                                                int j = 0;
                                                const char* foreign_field = NULL;
                                                osrfStringArray* ctx_array = osrfHashGet( fcontext, "context" );
                                                while ( (foreign_field = osrfStringArrayGetString( ctx_array, j++ )) ) {
                                                        osrfStringArrayAdd( context_org_array,
                                                                oilsFMGetStringConst( _fparam, foreign_field ));
                                                        osrfLogDebug( OSRF_LOG_MARK,
                                                                "adding foreign class %s field %s (value: %s) "
                                                                        "to the context org      list",
                                                                class_name,
                                                                foreign_field,
                                                                osrfStringArrayGetString(
                                                                        context_org_array, context_org_array->size - 1 )
                                                        );
                                                }

                                                jsonObjectFree( _fparam );
                                        }
                                }

                                osrfHashIteratorFree( class_itr );
                        }
                }

                jsonObjectFree( param );
        }

        const char* context_org = NULL;
        const char* perm = NULL;
        int OK = 0;

        if( permission->size == 0 ) {
            osrfLogDebug( OSRF_LOG_MARK, "No permission specified for this action, passing through" );
                OK = 1;
        }

        // For every combination of permission and context org unit: call a stored procedure
        // to determine if the user has this permission in the context of this org unit.
        // If the answer is yes at any point, then we're done, and the user has permission.
        // In other words permissions are additive.
        int i = 0;
        while( (perm = osrfStringArrayGetString(permission, i++)) ) {
                int j = 0;
                while( (context_org = osrfStringArrayGetString( context_org_array, j++ )) ) {
                        dbi_result result;

                        if( pkey_value ) {
                                osrfLogDebug(
                                        OSRF_LOG_MARK,
                                        "Checking object permission [%s] for user %d "
                                                        "on object %s (class %s) at org %d",
                                        perm,
                                        userid,
                                        pkey_value,
                                        osrfHashGet( class, "classname" ),
                                        atoi( context_org )
                                );

                                result = dbi_conn_queryf(
                                        writehandle,
                                        "SELECT permission.usr_has_object_perm(%d, '%s', '%s', '%s', %d) AS has_perm;",
                                        userid,
                                        perm,
                                        osrfHashGet( class, "classname" ),
                                        pkey_value,
                                        atoi( context_org )
                                );

                                if( result ) {
                                        osrfLogDebug(
                                                OSRF_LOG_MARK,
                                                "Received a result for object permission [%s] "
                                                                "for user %d on object %s (class %s) at org %d",
                                                perm,
                                                userid,
                                                pkey_value,
                                                osrfHashGet( class, "classname" ),
                                                atoi( context_org )
                                        );

                                        if( dbi_result_first_row( result )) {
                                                jsonObject* return_val = oilsMakeJSONFromResult( result );
                                                const char* has_perm = jsonObjectGetString(
                                                                jsonObjectGetKeyConst( return_val, "has_perm" ));

                                                osrfLogDebug(
                                                        OSRF_LOG_MARK,
                                                        "Status of object permission [%s] for user %d "
                                                                        "on object %s (class %s) at org %d is %s",
                                                        perm,
                                                        userid,
                                                        pkey_value,
                                                        osrfHashGet(class, "classname"),
                                                        atoi(context_org),
                                                        has_perm
                                                );

                                                if( *has_perm == 't' )
                                                        OK = 1;
                                                jsonObjectFree( return_val );
                                        }

                                        dbi_result_free( result );
                                        if( OK )
                                                break;
                                }
                        }

                        osrfLogDebug( OSRF_LOG_MARK,
                                        "Checking non-object permission [%s] for user %d at org %d",
                                        perm, userid, atoi(context_org) );
                        result = dbi_conn_queryf(
                                writehandle,
                                "SELECT permission.usr_has_perm(%d, '%s', %d) AS has_perm;",
                                userid,
                                perm,
                                atoi( context_org )
                        );

                        if( result ) {
                                osrfLogDebug( OSRF_LOG_MARK,
                                                "Received a result for permission [%s] for user %d at org %d",
                                                perm, userid, atoi( context_org ));
                                if( dbi_result_first_row( result )) {
                                        jsonObject* return_val = oilsMakeJSONFromResult( result );
                                        const char* has_perm = jsonObjectGetString(
                                                        jsonObjectGetKeyConst( return_val, "has_perm" ));
                                        osrfLogDebug( OSRF_LOG_MARK,
                                                        "Status of permission [%s] for user %d at org %d is [%s]",
                                                        perm, userid, atoi( context_org ), has_perm );
                                        if( *has_perm == 't' )
                                                OK = 1;
                                        jsonObjectFree( return_val );
                                }

                                dbi_result_free( result );
                                if( OK )
                                        break;
                        }

                }
                if( OK )
                        break;
        }

        osrfStringArrayFree( context_org_array );

        return OK;
}

/**
        @brief Look up the root of the org_unit tree.
        @param ctx Pointer to the method context.
        @return The id of the root org unit, as a character string.

        Query actor.org_unit where parent_ou is null, and return the id as a string.

        This function assumes that there is only one root org unit, i.e. that we
        have a single tree, not a forest.

        The calling code is responsible for freeing the returned string.
*/
static const char* org_tree_root( osrfMethodContext* ctx ) {

        static char cached_root_id[ 32 ] = "";  // extravagantly large buffer
        static time_t last_lookup_time = 0;
        time_t current_time = time( NULL );

        if( cached_root_id[ 0 ] && ( current_time - last_lookup_time < 3600 ) ) {
                // We successfully looked this up less than an hour ago.
                // It's not likely to have changed since then.
                return strdup( cached_root_id );
        }
        last_lookup_time = current_time;

        int err = 0;
        jsonObject* where_clause = single_hash( "parent_ou", NULL );
        jsonObject* result = doFieldmapperSearch(
                ctx, osrfHashGet( oilsIDL(), "aou" ), where_clause, NULL, &err );
        jsonObjectFree( where_clause );

        jsonObject* tree_top = jsonObjectGetIndex( result, 0 );

        if( !tree_top ) {
                jsonObjectFree( result );

                growing_buffer* msg = buffer_init( 128 );
                OSRF_BUFFER_ADD( msg, modulename );
                OSRF_BUFFER_ADD( msg,
                                ": Internal error, could not find the top of the org tree (parent_ou = NULL)" );

                char* m = buffer_release( msg );
                osrfAppSessionStatus( ctx->session,
                                OSRF_STATUS_INTERNALSERVERERROR, "osrfMethodException", ctx->request, m );
                free( m );

                cached_root_id[ 0 ] = '\0';
                return NULL;
        }

        const char* root_org_unit_id = oilsFMGetStringConst( tree_top, "id" );
        osrfLogDebug( OSRF_LOG_MARK, "Top of the org tree is %s", root_org_unit_id );

        strcpy( cached_root_id, root_org_unit_id );
        jsonObjectFree( result );
        return cached_root_id;
}

/**
        @brief Create a JSON_HASH with a single key/value pair.
        @param key The key of the key/value pair.
        @param value the value of the key/value pair.
        @return Pointer to a newly created jsonObject of type JSON_HASH.

        The value of the key/value is either a string or (if @a value is NULL) a null.
*/
static jsonObject* single_hash( const char* key, const char* value ) {
        // Sanity check
        if( ! key ) key = "";

        jsonObject* hash = jsonNewObjectType( JSON_HASH );
        jsonObjectSetKey( hash, key, jsonNewObject( value ) );
        return hash;
}


int doCreate( osrfMethodContext* ctx ) {
        if(osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
                return -1;
        }

        if( enforce_pcrud )
                timeout_needs_resetting = 1;

        osrfHash* meta = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );
        jsonObject* target = NULL;
        jsonObject* options = NULL;

        if( enforce_pcrud ) {
                target = jsonObjectGetIndex( ctx->params, 1 );
                options = jsonObjectGetIndex( ctx->params, 2 );
        } else {
                target = jsonObjectGetIndex( ctx->params, 0 );
                options = jsonObjectGetIndex( ctx->params, 1 );
        }

        if( !verifyObjectClass( ctx, target )) {
                osrfAppRespondComplete( ctx, NULL );
                return -1;
        }

        osrfLogDebug( OSRF_LOG_MARK, "Object seems to be of the correct type" );

        const char* trans_id = getXactId( ctx );
        if( !trans_id ) {
                osrfLogError( OSRF_LOG_MARK, "No active transaction -- required for CREATE" );

                osrfAppSessionStatus(
                        ctx->session,
                        OSRF_STATUS_BADREQUEST,
                        "osrfMethodException",
                        ctx->request,
                        "No active transaction -- required for CREATE"
                );
                osrfAppRespondComplete( ctx, NULL );
                return -1;
        }

        // The following test is harmless but redundant.  If a class is
        // readonly, we don't register a create method for it.
        if( str_is_true( osrfHashGet( meta, "readonly" ) ) ) {
                osrfAppSessionStatus(
                        ctx->session,
                        OSRF_STATUS_BADREQUEST,
                        "osrfMethodException",
                        ctx->request,
                        "Cannot INSERT readonly class"
                );
                osrfAppRespondComplete( ctx, NULL );
                return -1;
        }

        // Set the last_xact_id
        int index = oilsIDL_ntop( target->classname, "last_xact_id" );
        if( index > -1 ) {
                osrfLogDebug(OSRF_LOG_MARK, "Setting last_xact_id to %s on %s at position %d",
                        trans_id, target->classname, index);
                jsonObjectSetIndex( target, index, jsonNewObject( trans_id ));
        }

        osrfLogDebug( OSRF_LOG_MARK, "There is a transaction running..." );

        dbhandle = writehandle;

        osrfHash* fields = osrfHashGet( meta, "fields" );
        char* pkey       = osrfHashGet( meta, "primarykey" );
        char* seq        = osrfHashGet( meta, "sequence" );

        growing_buffer* table_buf = buffer_init( 128 );
        growing_buffer* col_buf   = buffer_init( 128 );
        growing_buffer* val_buf   = buffer_init( 128 );

        OSRF_BUFFER_ADD( table_buf, "INSERT INTO " );
        OSRF_BUFFER_ADD( table_buf, osrfHashGet( meta, "tablename" ));
        OSRF_BUFFER_ADD_CHAR( col_buf, '(' );
        buffer_add( val_buf,"VALUES (" );


        int first = 1;
        osrfHash* field = NULL;
        osrfHashIterator* field_itr = osrfNewHashIterator( fields );
        while( (field = osrfHashIteratorNext( field_itr ) ) ) {

                const char* field_name = osrfHashIteratorKey( field_itr );

                if( str_is_true( osrfHashGet( field, "virtual" ) ) )
                        continue;

                const jsonObject* field_object = oilsFMGetObject( target, field_name );

                char* value;
                if( field_object && field_object->classname ) {
                        value = oilsFMGetString(
                                field_object,
                                (char*)oilsIDLFindPath( "/%s/primarykey", field_object->classname )
                        );
                } else if( field_object && JSON_BOOL == field_object->type ) {
                        if( jsonBoolIsTrue( field_object ) )
                                value = strdup( "t" );
                        else
                                value = strdup( "f" );
                } else {
                        value = jsonObjectToSimpleString( field_object );
                }

                if( first ) {
                        first = 0;
                } else {
                        OSRF_BUFFER_ADD_CHAR( col_buf, ',' );
                        OSRF_BUFFER_ADD_CHAR( val_buf, ',' );
                }

                buffer_add( col_buf, field_name );

                if( !field_object || field_object->type == JSON_NULL ) {
                        buffer_add( val_buf, "DEFAULT" );

                } else if( !strcmp( get_primitive( field ), "number" )) {
                        const char* numtype = get_datatype( field );
                        if( !strcmp( numtype, "INT8" )) {
                                buffer_fadd( val_buf, "%lld", atoll( value ));

                        } else if( !strcmp( numtype, "INT" )) {
                                buffer_fadd( val_buf, "%d", atoi( value ));

                        } else if( !strcmp( numtype, "NUMERIC" )) {
                                buffer_fadd( val_buf, "%f", atof( value ));
                        }
                } else {
                        if( dbi_conn_quote_string( writehandle, &value )) {
                                OSRF_BUFFER_ADD( val_buf, value );

                        } else {
                                osrfLogError( OSRF_LOG_MARK, "%s: Error quoting string [%s]", modulename, value );
                                osrfAppSessionStatus(
                                        ctx->session,
                                        OSRF_STATUS_INTERNALSERVERERROR,
                                        "osrfMethodException",
                                        ctx->request,
                                        "Error quoting string -- please see the error log for more details"
                                );
                                free( value );
                                buffer_free( table_buf );
                                buffer_free( col_buf );
                                buffer_free( val_buf );
                                osrfAppRespondComplete( ctx, NULL );
                                return -1;
                        }
                }

                free( value );
        }

        osrfHashIteratorFree( field_itr );

        OSRF_BUFFER_ADD_CHAR( col_buf, ')' );
        OSRF_BUFFER_ADD_CHAR( val_buf, ')' );

        char* table_str = buffer_release( table_buf );
        char* col_str   = buffer_release( col_buf );
        char* val_str   = buffer_release( val_buf );
        growing_buffer* sql = buffer_init( 128 );
        buffer_fadd( sql, "%s %s %s;", table_str, col_str, val_str );
        free( table_str );
        free( col_str );
        free( val_str );

        char* query = buffer_release( sql );

        osrfLogDebug( OSRF_LOG_MARK, "%s: Insert SQL [%s]", modulename, query );

        jsonObject* obj = NULL;
        int rc = 0;

        dbi_result result = dbi_conn_query( writehandle, query );
        if( !result ) {
                obj = jsonNewObject( NULL );
                osrfLogError(
                        OSRF_LOG_MARK,
                        "%s ERROR inserting %s object using query [%s]",
                        modulename,
                        osrfHashGet(meta, "fieldmapper"),
                        query
                );
                osrfAppSessionStatus(
                        ctx->session,
                        OSRF_STATUS_INTERNALSERVERERROR,
                        "osrfMethodException",
                        ctx->request,
                        "INSERT error -- please see the error log for more details"
                );
                rc = -1;
        } else {

                char* id = oilsFMGetString( target, pkey );
                if( !id ) {
                        unsigned long long new_id = dbi_conn_sequence_last( writehandle, seq );
                        growing_buffer* _id = buffer_init( 10 );
                        buffer_fadd( _id, "%lld", new_id );
                        id = buffer_release( _id );
                }

                // Find quietness specification, if present
                const char* quiet_str = NULL;
                if( options ) {
                        const jsonObject* quiet_obj = jsonObjectGetKeyConst( options, "quiet" );
                        if( quiet_obj )
                                quiet_str = jsonObjectGetString( quiet_obj );
                }

                if( str_is_true( quiet_str )) {  // if quietness is specified
                        obj = jsonNewObject( id );
                }
                else {

                        // Fetch the row that we just inserted, so that we can return it to the client
                        jsonObject* where_clause = jsonNewObjectType( JSON_HASH );
                        jsonObjectSetKey( where_clause, pkey, jsonNewObject( id ));

                        int err = 0;
                        jsonObject* list = doFieldmapperSearch( ctx, meta, where_clause, NULL, &err );
                        if( err )
                                rc = -1;
                        else
                                obj = jsonObjectClone( jsonObjectGetIndex( list, 0 ));

                        jsonObjectFree( list );
                        jsonObjectFree( where_clause );
                }

                free( id );
        }

        free( query );
        osrfAppRespondComplete( ctx, obj );
        jsonObjectFree( obj );
        return rc;
}

/**
        @brief Implement the retrieve method.
        @param ctx Pointer to the method context.
        @param err Pointer through which to return an error code.
        @return If successful, a pointer to the result to be returned to the client;
        otherwise NULL.

        From the method's class, fetch a row with a specified value in the primary key.  This
        method relies on the database design convention that a primary key consists of a single
        column.

        Method parameters:
        - authkey (PCRUD only)
        - value of the primary key for the desired row, for building the WHERE clause
        - a JSON_HASH containing any other SQL clauses: select, join, etc.

        Return to client: One row from the query.
*/
int doRetrieve( osrfMethodContext* ctx ) {
        if(osrfMethodVerifyContext( ctx )) {
                osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
                return -1;
        }

        if( enforce_pcrud )
                timeout_needs_resetting = 1;

        int id_pos = 0;
        int order_pos = 1;

        if( enforce_pcrud ) {
                id_pos = 1;
                order_pos = 2;
        }

        // Get the class metadata
        osrfHash* class_def = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );

        // Get the value of the primary key, from a method parameter
        const jsonObject* id_obj = jsonObjectGetIndex( ctx->params, id_pos );

        osrfLogDebug(
                OSRF_LOG_MARK,
                "%s retrieving %s object with primary key value of %s",
                modulename,
                osrfHashGet( class_def, "fieldmapper" ),
                jsonObjectGetString( id_obj )
        );

        // Build a WHERE clause based on the key value
        jsonObject* where_clause = jsonNewObjectType( JSON_HASH );
        jsonObjectSetKey(
                where_clause,
                osrfHashGet( class_def, "primarykey" ),  // name of key column
                jsonObjectClone( id_obj )                // value of key column
        );

        jsonObject* rest_of_query = jsonObjectGetIndex( ctx->params, order_pos );

        // Do the query
        int err = 0;
        jsonObject* list = doFieldmapperSearch( ctx, class_def, where_clause, rest_of_query, &err );

        jsonObjectFree( where_clause );
        if( err ) {
                osrfAppRespondComplete( ctx, NULL );
                return -1;
        }

        jsonObject* obj = jsonObjectExtractIndex( list, 0 );
        jsonObjectFree( list );

        if( enforce_pcrud ) {
                if(!verifyObjectPCRUD( ctx, obj )) {
                        jsonObjectFree( obj );

                        growing_buffer* msg = buffer_init( 128 );
                        OSRF_BUFFER_ADD( msg, modulename );
                        OSRF_BUFFER_ADD( msg, ": Insufficient permissions to retrieve object" );

                        char* m = buffer_release( msg );
                        osrfAppSessionStatus( ctx->session, OSRF_STATUS_NOTALLOWED, "osrfMethodException",
                                        ctx->request, m );
                        free( m );

                        osrfAppRespondComplete( ctx, NULL );
                        return -1;
                }
        }

        osrfAppRespondComplete( ctx, obj );
        jsonObjectFree( obj );
        return 0;
}

/**
        @brief Translate a numeric value to a string representation for the database.
        @param field Pointer to the IDL field definition.
        @param value Pointer to a jsonObject holding the value of a field.
        @return Pointer to a newly allocated string.

        The input object is typically a JSON_NUMBER, but it may be a JSON_STRING as long as
        its contents are numeric.  A non-numeric string is likely to result in invalid SQL,
        or (what is worse) valid SQL that is wrong.

        If the datatype of the receiving field is not numeric, wrap the value in quotes.

        The calling code is responsible for freeing the resulting string by calling free().
*/
static char* jsonNumberToDBString( osrfHash* field, const jsonObject* value ) {
        growing_buffer* val_buf = buffer_init( 32 );
        const char* numtype = get_datatype( field );

        // For historical reasons the following contains cruft that could be cleaned up.
        if( !strncmp( numtype, "INT", 3 ) ) {
                if( value->type == JSON_NUMBER )
                        //buffer_fadd( val_buf, "%ld", (long)jsonObjectGetNumber(value) );
                        buffer_fadd( val_buf, jsonObjectGetString( value ) );
                else {
                        buffer_fadd( val_buf, jsonObjectGetString( value ) );
                }

        } else if( !strcmp( numtype, "NUMERIC" )) {
                if( value->type == JSON_NUMBER )
                        buffer_fadd( val_buf, jsonObjectGetString( value ));
                else {
                        buffer_fadd( val_buf, jsonObjectGetString( value ));
                }

        } else {
                // Presumably this was really intended to be a string, so quote it
                char* str = jsonObjectToSimpleString( value );
                if( dbi_conn_quote_string( dbhandle, &str )) {
                        OSRF_BUFFER_ADD( val_buf, str );
                        free( str );
                } else {
                        osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key string [%s]", modulename, str );
                        free( str );
                        buffer_free( val_buf );
                        return NULL;
                }
        }

        return buffer_release( val_buf );
}

static char* searchINPredicate( const char* class_alias, osrfHash* field,
                jsonObject* node, const char* op, osrfMethodContext* ctx ) {
        growing_buffer* sql_buf = buffer_init( 32 );

        buffer_fadd(
                sql_buf,
                "\"%s\".%s ",
                class_alias,
                osrfHashGet( field, "name" )
        );

        if( !op ) {
                buffer_add( sql_buf, "IN (" );
        } else if( !strcasecmp( op,"not in" )) {
                buffer_add( sql_buf, "NOT IN (" );
        } else {
                buffer_add( sql_buf, "IN (" );
        }

        if( node->type == JSON_HASH ) {
                // subquery predicate
                char* subpred = buildQuery( ctx, node, SUBSELECT );
                if( ! subpred ) {
                        buffer_free( sql_buf );
                        return NULL;
                }

                buffer_add( sql_buf, subpred );
                free( subpred );

        } else if( node->type == JSON_ARRAY ) {
                // literal value list
                int in_item_index = 0;
                int in_item_first = 1;
                const jsonObject* in_item;
                while( (in_item = jsonObjectGetIndex( node, in_item_index++ )) ) {

                        if( in_item_first )
                                in_item_first = 0;
                        else
                                buffer_add( sql_buf, ", " );

                        // Sanity check
                        if( in_item->type != JSON_STRING && in_item->type != JSON_NUMBER ) {
                                osrfLogError( OSRF_LOG_MARK,
                                                "%s: Expected string or number within IN list; found %s",
                                                modulename, json_type( in_item->type ) );
                                buffer_free( sql_buf );
                                return NULL;
                        }

                        // Append the literal value -- quoted if not a number
                        if( JSON_NUMBER == in_item->type ) {
                                char* val = jsonNumberToDBString( field, in_item );
                                OSRF_BUFFER_ADD( sql_buf, val );
                                free( val );

                        } else if( !strcmp( get_primitive( field ), "number" )) {
                                char* val = jsonNumberToDBString( field, in_item );
                                OSRF_BUFFER_ADD( sql_buf, val );
                                free( val );

                        } else {
                                char* key_string = jsonObjectToSimpleString( in_item );
                                if( dbi_conn_quote_string( dbhandle, &key_string )) {
                                        OSRF_BUFFER_ADD( sql_buf, key_string );
                                        free( key_string );
                                } else {
                                        osrfLogError( OSRF_LOG_MARK,
                                                        "%s: Error quoting key string [%s]", modulename, key_string );
                                        free( key_string );
                                        buffer_free( sql_buf );
                                        return NULL;
                                }
                        }
                }

                if( in_item_first ) {
                        osrfLogError(OSRF_LOG_MARK, "%s: Empty IN list", modulename );
                        buffer_free( sql_buf );
                        return NULL;
                }
        } else {
                osrfLogError( OSRF_LOG_MARK, "%s: Expected object or array for IN clause; found %s",
                        modulename, json_type( node->type ));
                buffer_free( sql_buf );
                return NULL;
        }

        OSRF_BUFFER_ADD_CHAR( sql_buf, ')' );

        return buffer_release( sql_buf );
}

// Receive a JSON_ARRAY representing a function call.  The first
// entry in the array is the function name.  The rest are parameters.
static char* searchValueTransform( const jsonObject* array ) {

        if( array->size < 1 ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Empty array for value transform", modulename );
                return NULL;
        }

        // Get the function name
        jsonObject* func_item = jsonObjectGetIndex( array, 0 );
        if( func_item->type != JSON_STRING ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Error: expected function name, found %s",
                        modulename, json_type( func_item->type ));
                return NULL;
        }

        growing_buffer* sql_buf = buffer_init( 32 );

        OSRF_BUFFER_ADD( sql_buf, jsonObjectGetString( func_item ) );
        OSRF_BUFFER_ADD( sql_buf, "( " );

        // Get the parameters
        int func_item_index = 1;   // We already grabbed the zeroth entry
        while( (func_item = jsonObjectGetIndex( array, func_item_index++ )) ) {

                // Add a separator comma, if we need one
                if( func_item_index > 2 )
                        buffer_add( sql_buf, ", " );

                // Add the current parameter
                if( func_item->type == JSON_NULL ) {
                        buffer_add( sql_buf, "NULL" );
                } else {
                        char* val = jsonObjectToSimpleString( func_item );
                        if( dbi_conn_quote_string( dbhandle, &val )) {
                                OSRF_BUFFER_ADD( sql_buf, val );
                                free( val );
                        } else {
                                osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key string [%s]",
                                        modulename, val );
                                buffer_free( sql_buf );
                                free( val );
                                return NULL;
                        }
                }
        }

        buffer_add( sql_buf, " )" );

        return buffer_release( sql_buf );
}

static char* searchFunctionPredicate( const char* class_alias, osrfHash* field,
                const jsonObject* node, const char* op ) {

        if( ! is_good_operator( op ) ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Invalid operator [%s]", modulename, op );
                return NULL;
        }

        char* val = searchValueTransform( node );
        if( !val )
                return NULL;

        growing_buffer* sql_buf = buffer_init( 32 );
        buffer_fadd(
                sql_buf,
                "\"%s\".%s %s %s",
                class_alias,
                osrfHashGet( field, "name" ),
                op,
                val
        );

        free( val );

        return buffer_release( sql_buf );
}

// class_alias is a class name or other table alias
// field is a field definition as stored in the IDL
// node comes from the method parameter, and may represent an entry in the SELECT list
static char* searchFieldTransform( const char* class_alias, osrfHash* field,
                const jsonObject* node ) {
        growing_buffer* sql_buf = buffer_init( 32 );

        const char* field_transform = jsonObjectGetString(
                jsonObjectGetKeyConst( node, "transform" ) );
        const char* transform_subcolumn = jsonObjectGetString(
                jsonObjectGetKeyConst( node, "result_field" ) );

        if( transform_subcolumn ) {
                if( ! is_identifier( transform_subcolumn ) ) {
                        osrfLogError( OSRF_LOG_MARK, "%s: Invalid subfield name: \"%s\"\n",
                                        modulename, transform_subcolumn );
                        buffer_free( sql_buf );
                        return NULL;
                }
                OSRF_BUFFER_ADD_CHAR( sql_buf, '(' );    // enclose transform in parentheses
        }

        if( field_transform ) {

                if( ! is_identifier( field_transform ) ) {
                        osrfLogError( OSRF_LOG_MARK, "%s: Expected function name, found \"%s\"\n",
                                        modulename, field_transform );
                        buffer_free( sql_buf );
                        return NULL;
                }

                if( obj_is_true( jsonObjectGetKeyConst( node, "distinct" ) ) ) {
                        buffer_fadd( sql_buf, "%s(DISTINCT \"%s\".%s",
                                field_transform, class_alias, osrfHashGet( field, "name" ));
                } else {
                        buffer_fadd( sql_buf, "%s(\"%s\".%s",
                                field_transform, class_alias, osrfHashGet( field, "name" ));
                }

                const jsonObject* array = jsonObjectGetKeyConst( node, "params" );

                if( array ) {
                        if( array->type != JSON_ARRAY ) {
                                osrfLogError( OSRF_LOG_MARK,
                                        "%s: Expected JSON_ARRAY for function params; found %s",
                                        modulename, json_type( array->type ) );
                                buffer_free( sql_buf );
                                return NULL;
                        }
                        int func_item_index = 0;
                        jsonObject* func_item;
                        while( (func_item = jsonObjectGetIndex( array, func_item_index++ ))) {

                                char* val = jsonObjectToSimpleString( func_item );

                                if( !val ) {
                                        buffer_add( sql_buf, ",NULL" );
                                } else if( dbi_conn_quote_string( dbhandle, &val )) {
                                        OSRF_BUFFER_ADD_CHAR( sql_buf, ',' );
                                        OSRF_BUFFER_ADD( sql_buf, val );
                                } else {
                                        osrfLogError( OSRF_LOG_MARK,
                                                        "%s: Error quoting key string [%s]", modulename, val );
                                        free( val );
                                        buffer_free( sql_buf );
                                        return NULL;
                                }
                                free( val );
                        }
                }

                buffer_add( sql_buf, " )" );

        } else {
                buffer_fadd( sql_buf, "\"%s\".%s", class_alias, osrfHashGet( field, "name" ));
        }

        if( transform_subcolumn )
                buffer_fadd( sql_buf, ").\"%s\"", transform_subcolumn );

        return buffer_release( sql_buf );
}

static char* searchFieldTransformPredicate( const ClassInfo* class_info, osrfHash* field,
                const jsonObject* node, const char* op ) {

        if( ! is_good_operator( op ) ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Error: Invalid operator %s", modulename, op );
                return NULL;
        }

        char* field_transform = searchFieldTransform( class_info->alias, field, node );
        if( ! field_transform )
                return NULL;
        char* value = NULL;
        int extra_parens = 0;   // boolean

        const jsonObject* value_obj = jsonObjectGetKeyConst( node, "value" );
        if( ! value_obj ) {
                value = searchWHERE( node, class_info, AND_OP_JOIN, NULL );
                if( !value ) {
                        osrfLogError( OSRF_LOG_MARK, "%s: Error building condition for field transform",
                                modulename );
                        free( field_transform );
                        return NULL;
                }
                extra_parens = 1;
        } else if( value_obj->type == JSON_ARRAY ) {
                value = searchValueTransform( value_obj );
                if( !value ) {
                        osrfLogError( OSRF_LOG_MARK,
                                "%s: Error building value transform for field transform", modulename );
                        free( field_transform );
                        return NULL;
                }
        } else if( value_obj->type == JSON_HASH ) {
                value = searchWHERE( value_obj, class_info, AND_OP_JOIN, NULL );
                if( !value ) {
                        osrfLogError( OSRF_LOG_MARK, "%s: Error building predicate for field transform",
                                modulename );
                        free( field_transform );
                        return NULL;
                }
                extra_parens = 1;
        } else if( value_obj->type == JSON_NUMBER ) {
                value = jsonNumberToDBString( field, value_obj );
        } else if( value_obj->type == JSON_NULL ) {
                osrfLogError( OSRF_LOG_MARK,
                        "%s: Error building predicate for field transform: null value", modulename );
                free( field_transform );
                return NULL;
        } else if( value_obj->type == JSON_BOOL ) {
                osrfLogError( OSRF_LOG_MARK,
                        "%s: Error building predicate for field transform: boolean value", modulename );
                free( field_transform );
                return NULL;
        } else {
                if( !strcmp( get_primitive( field ), "number") ) {
                        value = jsonNumberToDBString( field, value_obj );
                } else {
                        value = jsonObjectToSimpleString( value_obj );
                        if( !dbi_conn_quote_string( dbhandle, &value )) {
                                osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key string [%s]",
                                        modulename, value );
                                free( value );
                                free( field_transform );
                                return NULL;
                        }
                }
        }

        const char* left_parens  = "";
        const char* right_parens = "";

        if( extra_parens ) {
                left_parens  = "(";
                right_parens = ")";
        }

        growing_buffer* sql_buf = buffer_init( 32 );

        buffer_fadd(
                sql_buf,
                "%s%s %s %s %s %s%s",
                left_parens,
                field_transform,
                op,
                left_parens,
                value,
                right_parens,
                right_parens
        );

        free( value );
        free( field_transform );

        return buffer_release( sql_buf );
}

static char* searchSimplePredicate( const char* op, const char* class_alias,
                osrfHash* field, const jsonObject* node ) {

        if( ! is_good_operator( op ) ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Invalid operator [%s]", modulename, op );
                return NULL;
        }

        char* val = NULL;

        // Get the value to which we are comparing the specified column
        if( node->type != JSON_NULL ) {
                if( node->type == JSON_NUMBER ) {
                        val = jsonNumberToDBString( field, node );
                } else if( !strcmp( get_primitive( field ), "number" ) ) {
                        val = jsonNumberToDBString( field, node );
                } else {
                        val = jsonObjectToSimpleString( node );
                }
        }

        if( val ) {
                if( JSON_NUMBER != node->type && strcmp( get_primitive( field ), "number") ) {
                        // Value is not numeric; enclose it in quotes
                        if( !dbi_conn_quote_string( dbhandle, &val ) ) {
                                osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key string [%s]",
                                        modulename, val );
                                free( val );
                                return NULL;
                        }
                }
        } else {
                // Compare to a null value
                val = strdup( "NULL" );
                if( strcmp( op, "=" ))
                        op = "IS NOT";
                else
                        op = "IS";
        }

        growing_buffer* sql_buf = buffer_init( 32 );
        buffer_fadd( sql_buf, "\"%s\".%s %s %s", class_alias, osrfHashGet(field, "name"), op, val );
        char* pred = buffer_release( sql_buf );

        free( val );

        return pred;
}

static char* searchBETWEENPredicate( const char* class_alias,
                osrfHash* field, const jsonObject* node ) {

        const jsonObject* x_node = jsonObjectGetIndex( node, 0 );
        const jsonObject* y_node = jsonObjectGetIndex( node, 1 );

        if( NULL == y_node ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Not enough operands for BETWEEN operator", modulename );
                return NULL;
        }
        else if( NULL != jsonObjectGetIndex( node, 2 ) ) {
                osrfLogError( OSRF_LOG_MARK, "%s: Too many operands for BETWEEN operator", modulename );
                return NULL;
        }

        char* x_string;
        char* y_string;

        if( !strcmp( get_primitive( field ), "number") ) {
                x_string = jsonNumberToDBString( field, x_node );
                y_string = jsonNumberToDBString( field, y_node );

        } else {
                x_string = jsonObjectToSimpleString( x_node );
                y_string = jsonObjectToSimpleString( y_node );
                if( !(dbi_conn_quote_string( dbhandle, &x_string )
                        && dbi_conn_quote_string( dbhandle, &y_string )) ) {
                        osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key strings [%s] and [%s]",
                                        modulename, x_string, y_string );
                        free( x_string );
                        free( y_string );
                        return NULL;
                }
        }

        growing_buffer* sql_buf = buffer_init( 32 );
        buffer_fadd( sql_buf, "\"%s\".%s BETWEEN %s AND %s",
                        class_alias, osrfHashGet( field, "name" ), x_string, y_string );
        free( x_string );
        free( y_string );

        return buffer_release( sql_buf );
}

static char* searchPredicate( const ClassInfo* class_info, osrfHash* field,
                                                          jsonObject* node, osrfMethodContext* ctx ) {

        char* pred = NULL;
        if( node->type == JSON_ARRAY ) { // equality IN search
                pred = searchINPredicate( class_info->alias, field, node, NULL, ctx );
        } else if( node->type == JSON_HASH ) { // other search
                jsonIterator* pred_itr = jsonNewIterator( node );
                if( !jsonIteratorHasNext( pred_itr ) ) {
                        osrfLogError( OSRF_LOG_MARK, "%s: Empty predicate for field \"%s\"",
                                        modulename, osrfHashGet(field, "name" ));
                } else {
                        jsonObject* pred_node = jsonIteratorNext( pred_itr );

                        // Verify that there are no additional predicates
                        if( jsonIteratorHasNext( pred_itr ) ) {
                                osrfLogError( OSRF_LOG_MARK, "%s: Multiple predicates for field \"%s\"",
                                                modulename, osrfHashGet(field, "name" ));
                        } else if( !(strcasecmp( pred_itr->key,"between" )) )
                                pred = searchBETWEENPredicate( class_info->alias, field, pred_node );
                        else if( !(strcasecmp( pred_itr->key,"in" ))
                                        || !(strcasecmp( pred_itr->key,"not in" )) )
                                pred = searchINPredicate(
                                        class_info->alias, field, pred_node, pred_itr->key, ctx );
                        else if( pred_node->type == JSON_ARRAY )
                                pred = searchFunctionPredicate(
                                        class_info->alias, field, pred_node, pred_itr->key );
                        else if( pred_node->type == JSON_HASH )
                                pred = searchFieldTransformPredicate(
                                        class_info, field, pred_node, pred_itr->key );
                        else
                                pred = searchSimplePredicate( pred_itr->key, class_info->alias, field, pred_node );
                }
                jsonIteratorFree( pred_itr );

        } else if( node->type == JSON_NULL ) { // IS NULL search
                growing_buffer* _p = buffer_init( 64 );
                buffer_fadd(
                        _p,
                        "\"%s\".%s IS NULL",
                        class_info->class_name,
                        osrfHashGet( field, "name" )
                );
                pred = buffer_release( _p );
        } else { // equality search
                pred = searchSimplePredicate( "=", class_info->alias, field, node );
        }

        return pred;

}


/*

join : {
        acn : {
                field : record,
                fkey : id
                type : left
                filter_op : or
                filter : { ... },
                join : {
                        acp : {
                                field : call_number,
                                fkey : id,
                                filter : { ... },
                        },
                },
        },
        mrd : {
                field : record,
                type : inner
                fkey : id,
                filter : { ... },
        }
}

*/

static char* searchJOIN( const jsonObject* join_hash, const ClassInfo* left_info ) {

        const jsonObject* working_hash;
        jsonObject* freeable_hash = NULL;

        if( join_hash->type == JSON_HASH ) {
                working_hash = join_hash;
        } else if( join_hash->type == JSON_STRING ) {
                // turn it into a JSON_HASH by creating a wrapper
                // around a copy of the original
                const char* _tmp = jsonObjectGetString( join_hash );
                freeable_hash = jsonNewObjectType( JSON_HASH );
                jsonObjectSetKey( freeable_hash, _tmp, NULL );
                working_hash = freeable_hash;
        } else {
                osrfLogError(
                        OSRF_LOG_MARK,
                        "%s: JOIN failed; expected JSON object type not found",
                        modulename
                );
                return NULL;
        }

        growing_buffer* join_buf = buffer_init( 128 );
        const char* leftclass = left_info->class_name;

        jsonObject* snode = NULL;
        jsonIterator* search_itr = jsonNewIterator( working_hash );

        while ( (snode = jsonIteratorNext( search_itr )) ) {
                const char* right_alias = search_itr->key;
                const char* class =
                                jsonObjectGetString( jsonObjectGetKeyConst( snode, "class" ) );
                if( ! class )
                        class = right_alias;

                const ClassInfo* right_info = add_joined_class( right_alias, class );
                if( !right_info ) {
                        osrfLogError(
                                OSRF_LOG_MARK,
                                "%s: JOIN failed.  Class \"%s\" not resolved in IDL",
                                modulename,
                                search_itr->key
                        );
                        jsonIteratorFree( search_itr );
                        buffer_free( join_buf );
                        if( freeable_hash )
                                jsonObjectFree( freeable_hash );
                        return NULL;
                }
                osrfHash* links    = right_info->links;
                const char* table  = right_info->source_def;

                const char* fkey  = jsonObjectGetString( jsonObjectGetKeyConst( snode, "fkey" ) );
                const char* field = jsonObjectGetString( jsonObjectGetKeyConst( snode, "field" ) );

                if( field && !fkey ) {
                        // Look up the corresponding join column in the IDL.
                        // The link must be defined in the child table,
                        // and point to the right parent table.
                        osrfHash* idl_link = (osrfHash*) osrfHashGet( links, field );
                        const char* reltype = NULL;
                        const char* other_class = NULL;
                        reltype = osrfHashGet( idl_link, "reltype" );
                        if( reltype && strcmp( reltype, "has_many" ) )
                                other_class = osrfHashGet( idl_link, "class" );
                        if( other_class && !strcmp( other_class, leftclass ) )
                                fkey = osrfHashGet( idl_link, "key" );
                        if( !fkey ) {
                                osrfLogError(
                                        OSRF_LOG_MARK,
                                        "%s: JOIN failed.  No link defined from %s.%s to %s",
                                        modulename,
                                        class,
                                        field,
                                        leftclass
                                );
                                buffer_free( join_buf );
                                if( freeable_hash )
                                        jsonObjectFree( freeable_hash );
                                jsonIteratorFree( search_itr );
                                return NULL;
                        }

                } else if( !field && fkey ) {
                        // Look up the corresponding join column in the IDL.
                        // The link must be defined in the child table,
                        // and point to the right parent table.
                        osrfHash* left_links = left_info->links;
                        osrfHash* idl_link = (osrfHash*) osrfHashGet( left_links, fkey );
                        const char* reltype = NULL;
                        const char* other_class = NULL;
                        reltype = osrfHashGet( idl_link, "reltype" );
                        if( reltype && strcmp( reltype, "has_many" ) )
                                other_class = osrfHashGet( idl_link, "class" );
                        if( other_class && !strcmp( other_class, class ) )
                                field = osrfHashGet( idl_link, "key" );
                        if( !field ) {
                                osrfLogError(
                                        OSRF_LOG_MARK,
                                        "%s: JOIN failed.  No link defined from %s.%s to %s",
                                        modulename,
                                        leftclass,
                                        fkey,
                                        class
                                );
                                buffer_free( join_buf );
                                if( freeable_hash )
                                        jsonObjectFree( freeable_hash );
                                jsonIteratorFree( search_itr );
                                return NULL;
                        }

                } else if( !field && !fkey ) {
                        osrfHash* left_links = left_info->links;

                        // For each link defined for the left class:
                        // see if the link references the joined class
                        osrfHashIterator* itr = osrfNewHashIterator( left_links );
                        osrfHash* curr_link = NULL;
                        while( (curr_link = osrfHashIteratorNext( itr ) ) ) {
                                const char* other_class = osrfHashGet( curr_link, "class" );
                                if( other_class && !strcmp( other_class, class ) ) {

                                        // In the IDL, the parent class doesn't always know then names of the child
                                        // columns that are pointing to it, so don't use that end of the link
                                        const char* reltype = osrfHashGet( curr_link, "reltype" );
                                        if( reltype && strcmp( reltype, "has_many" ) ) {
                                                // Found a link between the classes
                                                fkey = osrfHashIteratorKey( itr );
                                                field = osrfHashGet( curr_link, "key" );
                                                break;
                                        }
                                }
                        }
                        osrfHashIteratorFree( itr );

                        if( !field || !fkey ) {
                                // Do another such search, with the classes reversed

                                // For each link defined for the joined class:
                                // see if the link references the left class
                                osrfHashIterator* itr = osrfNewHashIterator( links );
                                osrfHash* curr_link = NULL;
                                while( (curr_link = osrfHashIteratorNext( itr ) ) ) {
                                        const char* other_class = osrfHashGet( curr_link, "class" );
                                        if( other_class && !strcmp( other_class, leftclass ) ) {

                                                // In the IDL, the parent class doesn't know then names of the child