Open-ILS/src/c-apps/oils_cstore.c

   1 /**
   2         @file oils_cstore.c
   3         @brief As a server, perform database operations at the request of clients.
   4 */
   5
   6 #include <ctype.h>
   7 #include "opensrf/osrf_application.h"
   8 #include "opensrf/osrf_settings.h"
   9 #include "opensrf/osrf_message.h"
  10 #include "opensrf/utils.h"
  11 #include "opensrf/osrf_json.h"
  12 #include "opensrf/log.h"
  13 #include "openils/oils_utils.h"
  14 #include "openils/oils_constants.h"
  15 #include <dbi/dbi.h>
  16
  17 #include <time.h>
  18 #include <stdlib.h>
  19 #include <string.h>
  20 #include <unistd.h>
  21
  22 #ifdef RSTORE
  23 #  define MODULENAME "open-ils.reporter-store"
  24 #  define ENFORCE_PCRUD 0
  25 #else
  26 #  ifdef PCRUD
  27 #    define MODULENAME "open-ils.pcrud"
  28 #    define ENFORCE_PCRUD 1
  29 #  else
  30 #    define MODULENAME "open-ils.cstore"
  31 #    define ENFORCE_PCRUD 0
  32 #  endif
  33 #endif
  34
  35 // The next four macros are OR'd together as needed to form a set
  36 // of bitflags.  SUBCOMBO enables an extra pair of parentheses when
  37 // nesting one UNION, INTERSECT or EXCEPT inside another.
  38 // SUBSELECT tells us we're in a subquery, so don't add the
  39 // terminal semicolon yet.
  40 #define SUBCOMBO    8
  41 #define SUBSELECT   4
  42 #define DISABLE_I18N    2
  43 #define SELECT_DISTINCT 1
  44
  45 #define AND_OP_JOIN     0
  46 #define OR_OP_JOIN      1
  47
  48 struct ClassInfoStruct;
  49 typedef struct ClassInfoStruct ClassInfo;
  50
  51 #define ALIAS_STORE_SIZE 16
  52 #define CLASS_NAME_STORE_SIZE 16
  53
  54 struct ClassInfoStruct {
  55         char* alias;
  56         char* class_name;
  57         char* source_def;
  58         osrfHash* class_def;      // Points into IDL
  59         osrfHash* fields;         // Points into IDL
  60         osrfHash* links;          // Points into IDL
  61
  62         // The remaining members are private and internal.  Client code should not
  63         // access them directly.
  64
  65         ClassInfo* next;          // Supports linked list of joined classes
  66         int in_use;               // boolean
  67
  68         // We usually store the alias and class name in the following arrays, and
  69         // point the corresponding pointers at them.  When the string is too big
  70         // for the array (which will probably never happen in practice), we strdup it.
  71
  72         char alias_store[ ALIAS_STORE_SIZE + 1 ];
  73         char class_name_store[ CLASS_NAME_STORE_SIZE + 1 ];
  74 };
  75
  76 struct QueryFrameStruct;
  77 typedef struct QueryFrameStruct QueryFrame;
  78
  79 struct QueryFrameStruct {
  80         ClassInfo core;
  81         ClassInfo* join_list;  // linked list of classes joined to the core class
  82         QueryFrame* next;      // implements stack as linked list
  83         int in_use;            // boolean
  84 };
  85
  86 static int timeout_needs_resetting;
  87 static time_t time_next_reset;
  88
  89 int osrfAppChildInit();
  90 int osrfAppInitialize();
  91 void osrfAppChildExit();
  92
  93 static int verifyObjectClass ( osrfMethodContext*, const jsonObject* );
  94
  95 static void setXactId( osrfMethodContext* ctx );
  96 static inline const char* getXactId( osrfMethodContext* ctx );
  97 static inline void clearXactId( osrfMethodContext* ctx );
  98
  99 int beginTransaction ( osrfMethodContext* );
 100 int commitTransaction ( osrfMethodContext* );
 101 int rollbackTransaction ( osrfMethodContext* );
 102
 103 int setSavepoint ( osrfMethodContext* );
 104 int releaseSavepoint ( osrfMethodContext* );
 105 int rollbackSavepoint ( osrfMethodContext* );
 106
 107 int doJSONSearch ( osrfMethodContext* );
 108
 109 int dispatchCRUDMethod( osrfMethodContext* ctx );
 110 static int doSearch( osrfMethodContext* ctx );
 111 static int doIdList( osrfMethodContext* ctx );
 112 static int doCreate( osrfMethodContext* ctx );
 113 static int doRetrieve( osrfMethodContext* ctx );
 114 static int doUpdate( osrfMethodContext* ctx );
 115 static int doDelete ( osrfMethodContext* ctx );
 116 static jsonObject* doFieldmapperSearch ( osrfMethodContext* ctx, osrfHash* class_meta,
 117                 jsonObject* where_hash, jsonObject* query_hash, int* err );
 118 static jsonObject* oilsMakeFieldmapperFromResult( dbi_result, osrfHash* );
 119 static jsonObject* oilsMakeJSONFromResult( dbi_result );
 120
 121 static char* searchSimplePredicate ( const char* op, const char* class_alias,
 122                                 osrfHash* field, const jsonObject* node );
 123 static char* searchFunctionPredicate ( const char*, osrfHash*, const jsonObject*, const char* );
 124 static char* searchFieldTransform ( const char*, osrfHash*, const jsonObject*);
 125 static char* searchFieldTransformPredicate ( const ClassInfo*, osrfHash*, const jsonObject*, const char* );
 126 static char* searchBETWEENPredicate ( const char*, osrfHash*, const jsonObject* );
 127 static char* searchINPredicate ( const char*, osrfHash*,
 128                                                                  jsonObject*, const char*, osrfMethodContext* );
 129 static char* searchPredicate ( const ClassInfo*, osrfHash*, jsonObject*, osrfMethodContext* );
 130 static char* searchJOIN ( const jsonObject*, const ClassInfo* left_info );
 131 static char* searchWHERE ( const jsonObject*, const ClassInfo*, int, osrfMethodContext* );
 132 static char* buildSELECT ( jsonObject*, jsonObject*, osrfHash*, osrfMethodContext* );
 133 char* buildQuery( osrfMethodContext* ctx, jsonObject* query, int flags );
 134
 135 char* SELECT ( osrfMethodContext*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, int );
 136
 137 void userDataFree( void* );
 138 static void sessionDataFree( char*, void* );
 139 static char* getRelation( osrfHash* );
 140 static int str_is_true( const char* str );
 141 static int obj_is_true( const jsonObject* obj );
 142 static const char* json_type( int code );
 143 static const char* get_primitive( osrfHash* field );
 144 static const char* get_datatype( osrfHash* field );
 145 static int is_identifier( const char* s);
 146 static int is_good_operator( const char* op );
 147 static void pop_query_frame( void );
 148 static void push_query_frame( void );
 149 static int add_query_core( const char* alias, const char* class_name );
 150 static inline ClassInfo* search_alias( const char* target );
 151 static ClassInfo* search_all_alias( const char* target );
 152 static ClassInfo* add_joined_class( const char* alias, const char* classname );
 153 static void clear_query_stack( void );
 154
 155 static const jsonObject* verifyUserPCRUD( osrfMethodContext* );
 156 static int verifyObjectPCRUD( osrfMethodContext*, const jsonObject* );
 157 static char* org_tree_root( osrfMethodContext* ctx );
 158 static jsonObject* single_hash( const char* key, const char* value );
 159
 160 static int child_initialized = 0;   /* boolean */
 161
 162 static dbi_conn writehandle; /* our MASTER db connection */
 163 static dbi_conn dbhandle; /* our CURRENT db connection */
 164 //static osrfHash * readHandles;
 165 static jsonObject* const jsonNULL = NULL; //
 166 static int max_flesh_depth = 100;
 167
 168 // The following points to the top of a stack of QueryFrames.  It's a little
 169 // confusing because the top level of the query is at the bottom of the stack.
 170 static QueryFrame* curr_query = NULL;
 171
 172 //----------------------------------
 173
 174 int oilsExtendIDL( void );
 175
 176 static dbi_conn writehandle; /* our MASTER db connection */
 177 static dbi_conn dbhandle; /* our CURRENT db connection */
 178 //static osrfHash * readHandles;
 179
 180 static const int enforce_pcrud = ENFORCE_PCRUD;     // Boolean
 181 static const char modulename[] = MODULENAME;
 182
 183
 184 /**
 185         @brief Disconnect from the database.
 186
 187         This function is called when the server drone is about to terminate.
 188 */
 189 void osrfAppChildExit() {
 190         osrfLogDebug(OSRF_LOG_MARK, "Child is exiting, disconnecting from database...");
 191
 192         int same = 0;
 193         if (writehandle == dbhandle)
 194                 same = 1;
 195
 196         if (writehandle) {
 197                 dbi_conn_query(writehandle, "ROLLBACK;");
 198                 dbi_conn_close(writehandle);
 199                 writehandle = NULL;
 200         }
 201         if (dbhandle && !same)
 202                 dbi_conn_close(dbhandle);
 203
 204         // XXX add cleanup of readHandles whenever that gets used
 205
 206         return;
 207 }
 208
 209 /**
 210         @brief Initialize the application.
 211         @return Zero if successful, or non-zero if not.
 212
 213         Load the IDL file into an internal data structure for future reference.  Each non-virtual
 214         class in the IDL corresponds to a table or view in the database, or to a subquery defined
 215         in the IDL.  Ignore all virtual tables and virtual fields.
 216
 217         Register a number of methods, some of them general-purpose and others specific for
 218         particular classes.
 219
 220         The name of the application is given by the MODULENAME macro, whose value depends on
 221         conditional compilation.  The method names also incorporate MODULENAME, followed by a
 222         dot, as a prefix.  Some methods are registered or not registered depending on whether
 223         the PCRUD macro is defined, and on whether the IDL includes permacrud entries for the
 224         class and method.
 225
 226         The general-purpose methods are as follows (minus their MODULENAME prefixes):
 227
 228         - json_query (not registered for PCRUD)
 229         - transaction.begin
 230         - transaction.commit
 231         - transaction.rollback
 232         - savepoint.set
 233         - savepoint.release
 234         - savepoint.rollback
 235
 236         For each non-virtual class, create up to eight class-specific methods:
 237
 238         - create    (not for readonly classes)
 239         - retrieve
 240         - update    (not for readonly classes)
 241         - delete    (not for readonly classes
 242         - search    (atomic and non-atomic versions)
 243         - id_list   (atomic and non-atomic versions)
 244
 245         For PCRUD, the full method names follow the pattern "MODULENAME.method_type.classname".
 246         Otherwise they follow the pattern "MODULENAME.direct.XXX.method_type", where XXX is the
 247         fieldmapper name from the IDL, with every run of one or more consecutive colons replaced
 248         by a period.  In addition, the names of atomic methods have a suffix of ".atomic".
 249
 250         This function is called when the registering the application, and is executed by the
 251         listener before spawning the drones.
 252 */
 253 int osrfAppInitialize() {
 254
 255         osrfLogInfo(OSRF_LOG_MARK, "Initializing the CStore Server...");
 256         osrfLogInfo(OSRF_LOG_MARK, "Finding XML file...");
 257
 258         if (!oilsIDLInit( osrf_settings_host_value("/IDL") ))
 259                 return 1; /* return non-zero to indicate error */
 260
 261         growing_buffer* method_name = buffer_init(64);
 262
 263         if( ! enforce_pcrud ) {
 264                 // Generic search thingy (not for PCRUD)
 265                 buffer_add( method_name, modulename );
 266                 buffer_add( method_name, ".json_query" );
 267                 osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR( method_name ),
 268                         "doJSONSearch", "", 1, OSRF_METHOD_STREAMING );
 269         }
 270
 271         // first we register all the transaction and savepoint methods
 272         buffer_reset(method_name);
 273         OSRF_BUFFER_ADD(method_name, modulename );
 274         OSRF_BUFFER_ADD(method_name, ".transaction.begin");
 275         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR( method_name ),
 276                         "beginTransaction", "", 0, 0 );
 277
 278         buffer_reset(method_name);
 279         OSRF_BUFFER_ADD(method_name, modulename );
 280         OSRF_BUFFER_ADD(method_name, ".transaction.commit");
 281         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR(method_name),
 282                         "commitTransaction", "", 0, 0 );
 283
 284         buffer_reset(method_name);
 285         OSRF_BUFFER_ADD(method_name, modulename );
 286         OSRF_BUFFER_ADD(method_name, ".transaction.rollback");
 287         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR(method_name),
 288                         "rollbackTransaction", "", 0, 0 );
 289
 290         buffer_reset(method_name);
 291         OSRF_BUFFER_ADD(method_name, modulename );
 292         OSRF_BUFFER_ADD(method_name, ".savepoint.set");
 293         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR(method_name),
 294                         "setSavepoint", "", 1, 0 );
 295
 296         buffer_reset(method_name);
 297         OSRF_BUFFER_ADD(method_name, modulename );
 298         OSRF_BUFFER_ADD(method_name, ".savepoint.release");
 299         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR(method_name),
 300                         "releaseSavepoint", "", 1, 0 );
 301
 302         buffer_reset(method_name);
 303         OSRF_BUFFER_ADD(method_name, modulename );
 304         OSRF_BUFFER_ADD(method_name, ".savepoint.rollback");
 305         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR(method_name),
 306                         "rollbackSavepoint", "", 1, 0 );
 307
 308         static const char* global_method[] = {
 309                 "create",
 310                 "retrieve",
 311                 "update",
 312                 "delete",
 313                 "search",
 314                 "id_list"
 315         };
 316         const int global_method_count
 317                 = sizeof( global_method ) / sizeof ( global_method[0] );
 318
 319         unsigned long class_count = osrfHashGetCount( oilsIDL() );
 320         osrfLogDebug(OSRF_LOG_MARK, "%lu classes loaded", class_count );
 321         osrfLogDebug(OSRF_LOG_MARK,
 322                 "At most %lu methods will be generated",
 323                 (unsigned long) (class_count * global_method_count) );
 324
 325         osrfHashIterator* class_itr = osrfNewHashIterator( oilsIDL() );
 326         osrfHash* idlClass = NULL;
 327
 328         // For each class in the IDL...
 329         while( (idlClass = osrfHashIteratorNext( class_itr ) ) ) {
 330
 331                 const char* classname = osrfHashIteratorKey( class_itr );
 332                 osrfLogInfo(OSRF_LOG_MARK, "Generating class methods for %s", classname);
 333
 334                 if (!osrfStringArrayContains( osrfHashGet(idlClass, "controller"), modulename )) {
 335                         osrfLogInfo(OSRF_LOG_MARK, "%s is not listed as a controller for %s, moving on",
 336                                         modulename, classname);
 337                         continue;
 338                 }
 339
 340                 if ( str_is_true( osrfHashGet(idlClass, "virtual") ) ) {
 341                         osrfLogDebug(OSRF_LOG_MARK, "Class %s is virtual, skipping", classname );
 342                         continue;
 343                 }
 344
 345                 // Look up some other attributes of the current class
 346                 const char* idlClass_fieldmapper = osrfHashGet(idlClass, "fieldmapper");
 347                 if( !idlClass_fieldmapper ) {
 348                         osrfLogDebug( OSRF_LOG_MARK, "Skipping class \"%s\"; no fieldmapper in IDL",
 349                                         classname );
 350                         continue;
 351                 }
 352
 353                 osrfHash* idlClass_permacrud = NULL;
 354                 if( enforce_pcrud ) {
 355                         // For PCRUD, ignore classes with no permacrud section
 356                         idlClass_permacrud = osrfHashGet(idlClass, "permacrud");
 357                         if (!idlClass_permacrud) {
 358                                 osrfLogDebug( OSRF_LOG_MARK,
 359                                         "Skipping class \"%s\"; no permacrud in IDL", classname );
 360                                 continue;
 361                         }
 362                 }
 363
 364                 const char* readonly = osrfHashGet(idlClass, "readonly");
 365
 366                 int i;
 367                 for( i = 0; i < global_method_count; ++i ) {  // for each global method
 368                         const char* method_type = global_method[ i ];
 369                         osrfLogDebug(OSRF_LOG_MARK,
 370                                 "Using files to build %s class methods for %s", method_type, classname);
 371
 372                         if( enforce_pcrud ) {
 373                                 // Treat "id_list" or "search" as forms of "retrieve"
 374                                 const char* tmp_method = method_type;
 375                                 if ( *tmp_method == 'i' || *tmp_method == 's') {  // "id_list" or "search"
 376                                         tmp_method = "retrieve";
 377                                 }
 378                                 // Skip this method if there is no permacrud entry for it
 379                                 if (!osrfHashGet( idlClass_permacrud, tmp_method ))
 380                                         continue;
 381                         }
 382
 383                         // No create, update, or delete methods for a readonly class
 384                         if ( str_is_true( readonly )
 385                                 && ( *method_type == 'c' || *method_type == 'u' || *method_type == 'd') )
 386                                 continue;
 387
 388                         buffer_reset( method_name );
 389
 390                         // Build the method name
 391                         if( enforce_pcrud ) {
 392                                 // For PCRUD: MODULENAME.method_type.classname
 393                                 buffer_fadd(method_name, "%s.%s.%s", modulename, method_type, classname);
 394                         } else {
 395                                 // For non-PCRUD: MODULENAME.MODULENAME.direct.XXX.method_type
 396                                 // where XXX is the fieldmapper name from the IDL, with every run of
 397                                 // one or more consecutive colons replaced by a period.
 398                                 char* st_tmp = NULL;
 399                                 char* part = NULL;
 400                                 char* _fm = strdup( idlClass_fieldmapper );
 401                                 part = strtok_r(_fm, ":", &st_tmp);
 402
 403                                 buffer_fadd(method_name, "%s.direct.%s", modulename, part);
 404
 405                                 while ((part = strtok_r(NULL, ":", &st_tmp))) {
 406                                         OSRF_BUFFER_ADD_CHAR(method_name, '.');
 407                                         OSRF_BUFFER_ADD(method_name, part);
 408                                 }
 409                                 OSRF_BUFFER_ADD_CHAR(method_name, '.');
 410                                 OSRF_BUFFER_ADD(method_name, method_type);
 411                                 free(_fm);
 412                         }
 413
 414                         // For an id_list or search method we specify the OSRF_METHOD_STREAMING option.
 415                         // The consequence is that we implicitly create an atomic method in addition to
 416                         // the usual non-atomic method.
 417                         int flags = 0;
 418                         if (*method_type == 'i' || *method_type == 's') {  // id_list or search
 419                                 flags = flags | OSRF_METHOD_STREAMING;
 420                         }
 421
 422                         osrfHash* method_meta = osrfNewHash();
 423                         osrfHashSet( method_meta, idlClass, "class");
 424                         osrfHashSet( method_meta, buffer_data( method_name ), "methodname" );
 425                         osrfHashSet( method_meta, strdup(method_type), "methodtype" );
 426
 427                         // Register the method, with a pointer to an osrfHash to tell the method
 428                         // its name, type, and class.
 429                         osrfAppRegisterExtendedMethod(
 430                                 modulename,
 431                                 OSRF_BUFFER_C_STR( method_name ),
 432                                 "dispatchCRUDMethod",
 433                                 "",
 434                                 1,
 435                                 flags,
 436                                 (void*)method_meta
 437                         );
 438
 439                 } // end for each global method
 440         } // end for each class in IDL
 441
 442         buffer_free( method_name );
 443         osrfHashIteratorFree( class_itr );
 444
 445         return 0;
 446 }
 447
 448 /**
 449         @brief Get a table name, view name, or subquery for use in a FROM clause.
 450         @param class Pointer to the IDL class entry.
 451         @return A table name, a view name, or a subquery in parentheses.
 452
 453         In some cases the IDL defines a class, not with a table name or a view name, but with
 454         a SELECT statement, which may be used as a subquery.
 455 */
 456 static char* getRelation( osrfHash* class ) {
 457
 458         char* source_def = NULL;
 459         const char* tabledef = osrfHashGet(class, "tablename");
 460
 461         if ( tabledef ) {
 462                 source_def = strdup( tabledef );   // Return the name of a table or view
 463         } else {
 464                 tabledef = osrfHashGet( class, "source_definition" );
 465                 if( tabledef ) {
 466                         // Return a subquery, enclosed in parentheses
 467                         source_def = safe_malloc( strlen( tabledef ) + 3 );
 468                         source_def[ 0 ] = '(';
 469                         strcpy( source_def + 1, tabledef );
 470                         strcat( source_def, ")" );
 471                 } else {
 472                         // Not found: return an error
 473                         const char* classname = osrfHashGet( class, "classname" );
 474                         if( !classname )
 475                                 classname = "???";
 476                         osrfLogError(
 477                                 OSRF_LOG_MARK,
 478                                 "%s ERROR No tablename or source_definition for class \"%s\"",
 479                                 modulename,
 480                                 classname
 481                         );
 482                 }
 483         }
 484
 485         return source_def;
 486 }
 487
 488 /**
 489         @brief Initialize a server drone.
 490         @return Zero if successful, -1 if not.
 491
 492         Connect to the database.  For each non-virtual class in the IDL, execute a dummy "SELECT * "
 493         query to get the datatype of each column.  Record the datatypes in the loaded IDL.
 494
 495         This function is called by a server drone shortly after it is spawned by the listener.
 496 */
 497 int osrfAppChildInit() {
 498
 499         osrfLogDebug(OSRF_LOG_MARK, "Attempting to initialize libdbi...");
 500         dbi_initialize(NULL);
 501         osrfLogDebug(OSRF_LOG_MARK, "... libdbi initialized.");
 502
 503         char* driver = osrf_settings_host_value("/apps/%s/app_settings/driver", modulename );
 504         char* user   = osrf_settings_host_value("/apps/%s/app_settings/database/user", modulename );
 505         char* host   = osrf_settings_host_value("/apps/%s/app_settings/database/host", modulename );
 506         char* port   = osrf_settings_host_value("/apps/%s/app_settings/database/port", modulename );
 507         char* db     = osrf_settings_host_value("/apps/%s/app_settings/database/db", modulename );
 508         char* pw     = osrf_settings_host_value("/apps/%s/app_settings/database/pw", modulename );
 509         char* md     = osrf_settings_host_value("/apps/%s/app_settings/max_query_recursion",
 510                         modulename );
 511
 512         osrfLogDebug(OSRF_LOG_MARK, "Attempting to load the database driver [%s]...", driver);
 513         writehandle = dbi_conn_new(driver);
 514
 515         if(!writehandle) {
 516                 osrfLogError(OSRF_LOG_MARK, "Error loading database driver [%s]", driver);
 517                 return -1;
 518         }
 519         osrfLogDebug(OSRF_LOG_MARK, "Database driver [%s] seems OK", driver);
 520
 521         osrfLogInfo(OSRF_LOG_MARK, "%s connecting to database.  host=%s, "
 522                         "port=%s, user=%s, db=%s", modulename, host, port, user, db );
 523
 524         if(host) dbi_conn_set_option(writehandle, "host", host );
 525         if(port) dbi_conn_set_option_numeric( writehandle, "port", atoi(port) );
 526         if(user) dbi_conn_set_option(writehandle, "username", user);
 527         if(pw)   dbi_conn_set_option(writehandle, "password", pw );
 528         if(db)   dbi_conn_set_option(writehandle, "dbname", db );
 529
 530         if(md)                     max_flesh_depth = atoi(md);
 531         if(max_flesh_depth < 0)    max_flesh_depth = 1;
 532         if(max_flesh_depth > 1000) max_flesh_depth = 1000;
 533
 534         free(user);
 535         free(host);
 536         free(port);
 537         free(db);
 538         free(pw);
 539
 540         const char* err;
 541         if (dbi_conn_connect(writehandle) < 0) {
 542                 sleep(1);
 543                 if (dbi_conn_connect(writehandle) < 0) {
 544                         dbi_conn_error(writehandle, &err);
 545                         osrfLogError( OSRF_LOG_MARK, "Error connecting to database: %s", err);
 546                         return -1;
 547                 }
 548         }
 549
 550         osrfLogInfo(OSRF_LOG_MARK, "%s successfully connected to the database", modulename );
 551
 552         // Add datatypes from database to the fields in the IDL
 553         if( oilsExtendIDL() ) {
 554                 osrfLogError( OSRF_LOG_MARK, "Error extending the IDL" );
 555                 return -1;
 556         }
 557         else
 558                 return 0;
 559 }
 560
 561 /**
 562         @brief Add datatypes from the database to the fields in the IDL.
 563         @return Zero if successful, or 1 upon error.
 564
 565         For each relevant class in the IDL: ask the database for the datatype of every field.
 566         In particular, determine which fields are text fields and which fields are numeric
 567         fields, so that we know whether to enclose their values in quotes.
 568
 569         At this writing this function does not detect any errors, so it always returns zero.
 570 */
 571 int oilsExtendIDL( void ) {
 572         osrfHashIterator* class_itr = osrfNewHashIterator( oilsIDL() );
 573         osrfHash* class = NULL;
 574         growing_buffer* query_buf = buffer_init( 64 );
 575
 576         // For each class in the IDL...
 577         while( (class = osrfHashIteratorNext( class_itr ) ) ) {
 578                 const char* classname = osrfHashIteratorKey( class_itr );
 579                 osrfHash* fields = osrfHashGet( class, "fields" );
 580
 581                 // If the class is virtual, ignore it
 582                 if( str_is_true( osrfHashGet(class, "virtual") ) ) {
 583                         osrfLogDebug(OSRF_LOG_MARK, "Class %s is virtual, skipping", classname );
 584                         continue;
 585                 }
 586
 587                 char* tabledef = getRelation(class);
 588                 if( !tabledef )
 589                         continue;   // No such relation -- a query of it would be doomed to failure
 590
 591                 buffer_reset( query_buf );
 592                 buffer_fadd( query_buf, "SELECT * FROM %s AS x WHERE 1=0;", tabledef );
 593
 594                 free(tabledef);
 595
 596                 osrfLogDebug( OSRF_LOG_MARK, "%s Investigatory SQL = %s",
 597                                 modulename, OSRF_BUFFER_C_STR( query_buf ) );
 598
 599                 dbi_result result = dbi_conn_query( writehandle, OSRF_BUFFER_C_STR( query_buf ) );
 600                 if (result) {
 601
 602                         int columnIndex = 1;
 603                         const char* columnName;
 604                         osrfHash* _f;
 605                         while( (columnName = dbi_result_get_field_name(result, columnIndex)) ) {
 606
 607                                 osrfLogInternal( OSRF_LOG_MARK, "Looking for column named [%s]...",
 608                                                 columnName );
 609
 610                                 /* fetch the fieldmapper index */
 611                                 if( (_f = osrfHashGet(fields, columnName)) ) {
 612
 613                                         osrfLogDebug(OSRF_LOG_MARK, "Found [%s] in IDL hash...", columnName);
 614
 615                                         /* determine the field type and storage attributes */
 616
 617                                         switch( dbi_result_get_field_type_idx(result, columnIndex) ) {
 618
 619                                                 case DBI_TYPE_INTEGER : {
 620
 621                                                         if ( !osrfHashGet(_f, "primitive") )
 622                                                                 osrfHashSet(_f, "number", "primitive");
 623
 624                                                         int attr = dbi_result_get_field_attribs_idx(result, columnIndex);
 625                                                         if( attr & DBI_INTEGER_SIZE8 )
 626                                                                 osrfHashSet(_f, "INT8", "datatype");
 627                                                         else
 628                                                                 osrfHashSet(_f, "INT", "datatype");
 629                                                         break;
 630                                                 }
 631                                                 case DBI_TYPE_DECIMAL :
 632                                                         if ( !osrfHashGet(_f, "primitive") )
 633                                                                 osrfHashSet(_f, "number", "primitive");
 634
 635                                                         osrfHashSet(_f,"NUMERIC", "datatype");
 636                                                         break;
 637
 638                                                 case DBI_TYPE_STRING :
 639                                                         if ( !osrfHashGet(_f, "primitive") )
 640                                                                 osrfHashSet(_f,"string", "primitive");
 641
 642                                                         osrfHashSet(_f,"TEXT", "datatype");
 643                                                         break;
 644
 645                                                 case DBI_TYPE_DATETIME :
 646                                                         if ( !osrfHashGet(_f, "primitive") )
 647                                                                 osrfHashSet(_f,"string", "primitive");
 648
 649                                                         osrfHashSet(_f,"TIMESTAMP", "datatype");
 650                                                         break;
 651
 652                                                 case DBI_TYPE_BINARY :
 653                                                         if ( !osrfHashGet(_f, "primitive") )
 654                                                                 osrfHashSet(_f,"string", "primitive");
 655
 656                                                         osrfHashSet(_f,"BYTEA", "datatype");
 657                                         }
 658
 659                                         osrfLogDebug(
 660                                                 OSRF_LOG_MARK,
 661                                                 "Setting [%s] to primitive [%s] and datatype [%s]...",
 662                                                 columnName,
 663                                                 osrfHashGet(_f, "primitive"),
 664                                                 osrfHashGet(_f, "datatype")
 665                                         );
 666                                 }
 667                                 ++columnIndex;
 668                         } // end while loop for traversing columns of result
 669                         dbi_result_free(result);
 670                 } else {
 671                         osrfLogDebug(OSRF_LOG_MARK, "No data found for class [%s]...", classname);
 672                 }
 673         } // end for each class in IDL
 674
 675         buffer_free( query_buf );
 676         osrfHashIteratorFree( class_itr );
 677         child_initialized = 1;
 678         return 0;
 679 }
 680
 681 /**
 682         @brief Install a database driver.
 683         @param conn Pointer to a database driver.
 684
 685         The driver is used to process quoted strings correctly.
 686
 687         This function is a sleazy hack, intended @em only for testing and debugging without
 688         actually connecting to a database. Any real server process should initialize the
 689         database connection by calling osrfAppChildInit().
 690 */
 691 void set_cstore_dbi_conn( dbi_conn conn ) {
 692         dbhandle = writehandle = conn;
 693 }
 694
 695 /**
 696         @brief Free an osrfHash that stores a transaction ID.
 697         @param blob A pointer to the osrfHash to be freed, cast to a void pointer.
 698
 699         This function is a callback, to be called by the application session when it ends.
 700         The application session stores the osrfHash via an opaque pointer.
 701
 702         If the osrfHash contains an entry for the key "xact_id", it means that an
 703         uncommitted transaction is pending.  Roll it back.
 704 */
 705 void userDataFree( void* blob ) {
 706         osrfHash* hash = (osrfHash*) blob;
 707         if( osrfHashGet( hash, "xact_id" ) && writehandle )
 708                 dbi_conn_query( writehandle, "ROLLBACK;" );
 709
 710         osrfHashFree( hash );
 711 }
 712
 713 /**
 714         @name Managing session data
 715         @brief Maintain data stored via the userData pointer of the application session.
 716
 717         Currently, session-level data is stored in an osrfHash.  Other arrangements are
 718         possible, and some would be more efficient.  The application session calls a
 719         callback function to free userData before terminating.
 720
 721         Currently, the only data we store at the session level is the transaction id.  By this
 722         means we can ensure that any pending transactions are rolled back before the application
 723         session terminates.
 724 */
 725 /*@{*/
 726
 727 /**
 728         @brief Free an item in the application session's userData.
 729         @param key The name of a key for an osrfHash.
 730         @param item An opaque pointer to the item associated with the key.
 731
 732         We store an osrfHash as userData with the application session, and arrange (by
 733         installing userDataFree() as a different callback) for the session to free that
 734         osrfHash before terminating.
 735
 736         This function is a callback for freeing items in the osrfHash.  Currently we store
 737         two things:
 738         - Transaction id of a pending transaction; a character string.  Key: "xact_id".
 739         - Authkey; a character string.  Key: "authkey".
 740         - User object from the authentication server; a jsonObject.  Key: "user_login".
 741
 742         If we ever store anything else in userData, we will need to revisit this function so
 743         that it will free whatever else needs freeing.
 744 */
 745 static void sessionDataFree( char* key, void* item ) {
 746         if ( !strcmp( key, "xact_id" )
 747              || !strcmp( key, "authkey" ) ) {
 748                 free( item );
 749         } else if( !strcmp( key, "user_login" ) )
 750                 jsonObjectFree( (jsonObject*) item );
 751 }
 752
 753 /**
 754         @brief Save a transaction id.
 755         @param ctx Pointer to the method context.
 756
 757         Save the session_id of the current application session as a transaction id.
 758 */
 759 static void setXactId( osrfMethodContext* ctx ) {
 760         if( ctx && ctx->session ) {
 761                 osrfAppSession* session = ctx->session;
 762
 763                 osrfHash* cache = session->userData;
 764
 765                 // If the session doesn't already have a hash, create one.  Make sure
 766                 // that the application session frees the hash when it terminates.
 767                 if( NULL == cache ) {
 768                         session->userData = cache = osrfNewHash();
 769                         osrfHashSetCallback( cache, &sessionDataFree );
 770                         ctx->session->userDataFree = &userDataFree;
 771                 }
 772
 773                 // Save the transaction id in the hash, with the key "xact_id"
 774                 osrfHashSet( cache, strdup( session->session_id ), "xact_id" );
 775         }
 776 }
 777
 778 /**
 779         @brief Get the transaction ID for the current transaction, if any.
 780         @param ctx Pointer to the method context.
 781         @return Pointer to the transaction ID.
 782
 783         The return value points to an internal buffer, and will become invalid upon issuing
 784         a commit or rollback.
 785 */
 786 static inline const char* getXactId( osrfMethodContext* ctx ) {
 787         if( ctx && ctx->session && ctx->session->userData )
 788                 return osrfHashGet( (osrfHash*) ctx->session->userData, "xact_id" );
 789         else
 790                 return NULL;
 791 }
 792
 793 /**
 794         @brief Clear the current transaction id.
 795         @param ctx Pointer to the method context.
 796 */
 797 static inline void clearXactId( osrfMethodContext* ctx ) {
 798         if( ctx && ctx->session && ctx->session->userData )
 799                 osrfHashRemove( ctx->session->userData, "xact_id" );
 800 }
 801 /*@}*/
 802
 803 /**
 804         @brief Save the user's login in the userData for the current application session.
 805         @param ctx Pointer to the method context.
 806         @param user_login Pointer to the user login object to be cached (we cache the original,
 807         not a copy of it).
 808
 809         If @a user_login is NULL, remove the user login if one is already cached.
 810 */
 811 static void setUserLogin( osrfMethodContext* ctx, jsonObject* user_login ) {
 812         if( ctx && ctx->session ) {
 813                 osrfAppSession* session = ctx->session;
 814
 815                 osrfHash* cache = session->userData;
 816
 817                 // If the session doesn't already have a hash, create one.  Make sure
 818                 // that the application session frees the hash when it terminates.
 819                 if( NULL == cache ) {
 820                         session->userData = cache = osrfNewHash();
 821                         osrfHashSetCallback( cache, &sessionDataFree );
 822                         ctx->session->userDataFree = &userDataFree;
 823                 }
 824
 825                 if( user_login )
 826                         osrfHashSet( cache, user_login, "user_login" );
 827                 else
 828                         osrfHashRemove( cache, "user_login" );
 829         }
 830 }
 831
 832 /**
 833         @brief Get the user login object for the current application session, if any.
 834         @param ctx Pointer to the method context.
 835         @return Pointer to the user login object if found; otherwise NULL.
 836
 837         The user login object was returned from the authentication server, and then cached so
 838         we don't have to call the authentication server again for the same user.
 839 */
 840 static const jsonObject* getUserLogin( osrfMethodContext* ctx ) {
 841         if( ctx && ctx->session && ctx->session->userData )
 842                 return osrfHashGet( (osrfHash*) ctx->session->userData, "user_login" );
 843         else
 844                 return NULL;
 845 }
 846
 847 /**
 848         @brief Save a copy of an authkey in the userData of the current application session.
 849         @param ctx Pointer to the method context.
 850         @param authkey The authkey to be saved.
 851
 852         If @a authkey is NULL, remove the authkey if one is already cached.
 853 */
 854 static void setAuthkey( osrfMethodContext* ctx, const char* authkey ) {
 855         if( ctx && ctx->session && authkey ) {
 856                 osrfAppSession* session = ctx->session;
 857                 osrfHash* cache = session->userData;
 858
 859                 // If the session doesn't already have a hash, create one.  Make sure
 860                 // that the application session frees the hash when it terminates.
 861                 if( NULL == cache ) {
 862                         session->userData = cache = osrfNewHash();
 863                         osrfHashSetCallback( cache, &sessionDataFree );
 864                         ctx->session->userDataFree = &userDataFree;
 865                 }
 866
 867                 // Save the transaction id in the hash, with the key "xact_id"
 868                 if( authkey && *authkey )
 869                         osrfHashSet( cache, strdup( authkey ), "authkey" );
 870                 else
 871                         osrfHashRemove( cache, "authkey" );
 872         }
 873 }
 874
 875 /**
 876         @brief Reset the login timeout.
 877         @param authkey The authentication key for the current login session.
 878         @param now The current time.
 879         @return Zero if successful, or 1 if not.
 880
 881         Tell the authentication server to reset the timeout so that the login session won't
 882         expire for a while longer.
 883
 884         We could dispense with the @a now parameter by calling time().  But we just called
 885         time() in order to decide whether to reset the timeout, so we might as well reuse
 886         the result instead of calling time() again.
 887 */
 888 static int reset_timeout( const char* authkey, time_t now ) {
 889         jsonObject* auth_object = jsonNewObject( authkey );
 890
 891         // Ask the authentication server to reset the timeout.  It returns an event
 892         // indicating success or failure.
 893         jsonObject* result = oilsUtilsQuickReq( "open-ils.auth",
 894                 "open-ils.auth.session.reset_timeout", auth_object );
 895         jsonObjectFree(auth_object);
 896
 897         if( !result || result->type != JSON_HASH ) {
 898                 osrfLogError( OSRF_LOG_MARK,
 899                          "Unexpected object type receieved from open-ils.auth.session.reset_timeout" );
 900                 jsonObjectFree( result );
 901                 return 1;       // Not the right sort of object returned
 902         }
 903
 904         const jsonObject* ilsevent = jsonObjectGetKey( result, "ilsevent" );
 905         if( !ilsevent || ilsevent->type != JSON_NUMBER ) {
 906                 osrfLogError( OSRF_LOG_MARK, "ilsevent is absent or malformed" );
 907                 jsonObjectFree( result );
 908                 return 1;    // Return code from method not available
 909         }
 910
 911         if( jsonObjectGetNumber( ilsevent ) != 0.0 ){
 912                 const char* desc = jsonObjectGetString( jsonObjectGetKey( result, "desc" ) );
 913                 if( !desc )
 914                         desc = "(No reason available)";    // failsafe; shouldn't happen
 915                 osrfLogInfo( OSRF_LOG_MARK, "Failure to reset timeout: %s", desc );
 916                 jsonObjectFree( result );
 917                 return 1;
 918         }
 919
 920         // Revise our local proxy for the timeout deadline
 921         // by a smallish fraction of the timeout interval
 922         const char* timeout = jsonObjectGetString( jsonObjectGetKey( result, "payload" ) );
 923         if( !timeout )
 924                 timeout = "1";   // failsafe; shouldn't happen
 925         time_next_reset = now + atoi( timeout ) / 15;
 926
 927         jsonObjectFree( result );
 928         return 0;     // Successfully reset timeout
 929 }
 930
 931 /**
 932         @brief Get the authkey string for the current application session, if any.
 933         @param ctx Pointer to the method context.
 934         @return Pointer to the cached authkey if found; otherwise NULL.
 935
 936         If present, the authkey string was cached from a previous method call.
 937 */
 938 static const char* getAuthkey( osrfMethodContext* ctx ) {
 939         if( ctx && ctx->session && ctx->session->userData ) {
 940                 const char* authkey = osrfHashGet( (osrfHash*) ctx->session->userData, "authkey" );
 941
 942                 // Possibly reset the authentication timeout to keep the login alive.  We do so
 943                 // no more than once per method call, and not at all if it has been only a short
 944                 // time since the last reset.
 945
 946                 // Here we reset explicitly, if at all.  We also implicitly reset the timeout
 947                 // whenever we call the "open-ils.auth.session.retrieve" method.
 948                 if( timeout_needs_resetting ) {
 949                         time_t now = time( NULL );
 950                         if( now >= time_next_reset && reset_timeout( authkey, now ) )
 951                                 authkey = NULL;    // timeout has apparently expired already
 952                 }
 953
 954                 timeout_needs_resetting = 0;
 955                 return authkey;
 956         }
 957         else
 958                 return NULL;
 959 }
 960
 961 /**
 962         @brief Implement the transaction.begin method.
 963         @param ctx Pointer to the method context.
 964         @return Zero if successful, or -1 upon error.
 965
 966         Start a transaction.  Save a transaction ID for future reference.
 967
 968         Method parameters:
 969         - authkey (PCRUD only)
 970
 971         Return to client: Transaction ID
 972 */
 973 int beginTransaction ( osrfMethodContext* ctx ) {
 974         if(osrfMethodVerifyContext( ctx )) {
 975                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
 976                 return -1;
 977         }
 978
 979         if( enforce_pcrud ) {
 980                 timeout_needs_resetting = 1;
 981                 const jsonObject* user = verifyUserPCRUD( ctx );
 982                 if (!user)
 983                         return -1;
 984         }
 985
 986         dbi_result result = dbi_conn_query(writehandle, "START TRANSACTION;");
 987         if (!result) {
 988                 osrfLogError(OSRF_LOG_MARK, "%s: Error starting transaction", modulename );
 989                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
 990                                 "osrfMethodException", ctx->request, "Error starting transaction" );
 991                 return -1;
 992         } else {
 993                 setXactId( ctx );
 994                 jsonObject* ret = jsonNewObject( getXactId( ctx ) );
 995                 osrfAppRespondComplete( ctx, ret );
 996                 jsonObjectFree(ret);
 997         }
 998         return 0;
 999 }
1000
1001 /**
1002         @brief Implement the savepoint.set method.
1003         @param ctx Pointer to the method context.
1004         @return Zero if successful, or -1 if not.
1005
1006         Issue a SAVEPOINT to the database server.
1007
1008         Method parameters:
1009         - authkey (PCRUD only)
1010         - savepoint name
1011
1012         Return to client: Savepoint name
1013 */
1014 int setSavepoint ( osrfMethodContext* ctx ) {
1015         if(osrfMethodVerifyContext( ctx )) {
1016                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
1017                 return -1;
1018         }
1019
1020         int spNamePos = 0;
1021         if( enforce_pcrud ) {
1022                 spNamePos = 1;
1023                 timeout_needs_resetting = 1;
1024                 const jsonObject* user = verifyUserPCRUD( ctx );
1025                 if (!user)
1026                         return -1;
1027         }
1028
1029         // Verify that a transaction is pending
1030         const char* trans_id = getXactId( ctx );
1031         if( NULL == trans_id ) {
1032                 osrfAppSessionStatus(
1033                         ctx->session,
1034                         OSRF_STATUS_INTERNALSERVERERROR,
1035                         "osrfMethodException",
1036                         ctx->request,
1037                         "No active transaction -- required for savepoints"
1038                 );
1039                 return -1;
1040         }
1041
1042         // Get the savepoint name from the method params
1043         const char* spName = jsonObjectGetString( jsonObjectGetIndex(ctx->params, spNamePos) );
1044
1045         dbi_result result = dbi_conn_queryf(writehandle, "SAVEPOINT \"%s\";", spName);
1046         if (!result) {
1047                 osrfLogError(
1048                         OSRF_LOG_MARK,
1049                         "%s: Error creating savepoint %s in transaction %s",
1050                         modulename,
1051                         spName,
1052                         trans_id
1053                 );
1054                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1055                                 "osrfMethodException", ctx->request, "Error creating savepoint" );
1056                 return -1;
1057         } else {
1058                 jsonObject* ret = jsonNewObject(spName);
1059                 osrfAppRespondComplete( ctx, ret );
1060                 jsonObjectFree(ret);
1061         }
1062         return 0;
1063 }
1064
1065 /**
1066         @brief Implement the savepoint.release method.
1067         @param ctx Pointer to the method context.
1068         @return Zero if successful, or -1 if not.
1069
1070         Issue a RELEASE SAVEPOINT to the database server.
1071
1072         Method parameters:
1073         - authkey (PCRUD only)
1074         - savepoint name
1075
1076         Return to client: Savepoint name
1077 */
1078 int releaseSavepoint ( osrfMethodContext* ctx ) {
1079         if(osrfMethodVerifyContext( ctx )) {
1080                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
1081                 return -1;
1082         }
1083
1084         int spNamePos = 0;
1085         if( enforce_pcrud ) {
1086                 spNamePos = 1;
1087                 timeout_needs_resetting = 1;
1088                 const jsonObject* user = verifyUserPCRUD( ctx );
1089                 if (!user)
1090                         return -1;
1091         }
1092
1093         // Verify that a transaction is pending
1094         const char* trans_id = getXactId( ctx );
1095         if( NULL == trans_id ) {
1096                 osrfAppSessionStatus(
1097                         ctx->session,
1098                         OSRF_STATUS_INTERNALSERVERERROR,
1099                         "osrfMethodException",
1100                         ctx->request,
1101                         "No active transaction -- required for savepoints"
1102                 );
1103                 return -1;
1104         }
1105
1106         // Get the savepoint name from the method params
1107         const char* spName = jsonObjectGetString( jsonObjectGetIndex(ctx->params, spNamePos) );
1108
1109         dbi_result result = dbi_conn_queryf(writehandle, "RELEASE SAVEPOINT \"%s\";", spName);
1110         if (!result) {
1111                 osrfLogError(
1112                         OSRF_LOG_MARK,
1113                         "%s: Error releasing savepoint %s in transaction %s",
1114                         modulename,
1115                         spName,
1116                         trans_id
1117                 );
1118                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1119                                 "osrfMethodException", ctx->request, "Error releasing savepoint" );
1120                 return -1;
1121         } else {
1122                 jsonObject* ret = jsonNewObject(spName);
1123                 osrfAppRespondComplete( ctx, ret );
1124                 jsonObjectFree(ret);
1125         }
1126         return 0;
1127 }
1128
1129 /**
1130         @brief Implement the savepoint.rollback method.
1131         @param ctx Pointer to the method context.
1132         @return Zero if successful, or -1 if not.
1133
1134         Issue a ROLLBACK TO SAVEPOINT to the database server.
1135
1136         Method parameters:
1137         - authkey (PCRUD only)
1138         - savepoint name
1139
1140         Return to client: Savepoint name
1141 */
1142 int rollbackSavepoint ( osrfMethodContext* ctx ) {
1143         if(osrfMethodVerifyContext( ctx )) {
1144                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
1145                 return -1;
1146         }
1147
1148         int spNamePos = 0;
1149         if( enforce_pcrud ) {
1150                 spNamePos = 1;
1151                 timeout_needs_resetting = 1;
1152                 const jsonObject* user = verifyUserPCRUD( ctx );
1153                 if (!user)
1154                         return -1;
1155         }
1156
1157         // Verify that a transaction is pending
1158         const char* trans_id = getXactId( ctx );
1159         if( NULL == trans_id ) {
1160                 osrfAppSessionStatus(
1161                         ctx->session,
1162                         OSRF_STATUS_INTERNALSERVERERROR,
1163                         "osrfMethodException",
1164                         ctx->request,
1165                         "No active transaction -- required for savepoints"
1166                 );
1167                 return -1;
1168         }
1169
1170         // Get the savepoint name from the method params
1171         const char* spName = jsonObjectGetString( jsonObjectGetIndex(ctx->params, spNamePos) );
1172
1173         dbi_result result = dbi_conn_queryf(writehandle, "ROLLBACK TO SAVEPOINT \"%s\";", spName);
1174         if (!result) {
1175                 osrfLogError(
1176                         OSRF_LOG_MARK,
1177                         "%s: Error rolling back savepoint %s in transaction %s",
1178                         modulename,
1179                         spName,
1180                         trans_id
1181                 );
1182                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1183                                 "osrfMethodException", ctx->request, "Error rolling back savepoint" );
1184                 return -1;
1185         } else {
1186                 jsonObject* ret = jsonNewObject(spName);
1187                 osrfAppRespondComplete( ctx, ret );
1188                 jsonObjectFree(ret);
1189         }
1190         return 0;
1191 }
1192
1193 /**
1194         @brief Implement the transaction.commit method.
1195         @param ctx Pointer to the method context.
1196         @return Zero if successful, or -1 if not.
1197
1198         Issue a COMMIT to the database server.
1199
1200         Method parameters:
1201         - authkey (PCRUD only)
1202
1203         Return to client: Transaction ID.
1204 */
1205 int commitTransaction ( osrfMethodContext* ctx ) {
1206         if(osrfMethodVerifyContext( ctx )) {
1207                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
1208                 return -1;
1209         }
1210
1211         if( enforce_pcrud ) {
1212                 timeout_needs_resetting = 1;
1213                 const jsonObject* user = verifyUserPCRUD( ctx );
1214                 if (!user)
1215                         return -1;
1216         }
1217
1218         // Verify that a transaction is pending
1219         const char* trans_id = getXactId( ctx );
1220         if( NULL == trans_id ) {
1221                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1222                                 "osrfMethodException", ctx->request, "No active transaction to commit" );
1223                 return -1;
1224         }
1225
1226         dbi_result result = dbi_conn_query(writehandle, "COMMIT;");
1227         if (!result) {
1228                 osrfLogError(OSRF_LOG_MARK, "%s: Error committing transaction", modulename );
1229                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1230                                 "osrfMethodException", ctx->request, "Error committing transaction" );
1231                 return -1;
1232         } else {
1233                 jsonObject* ret = jsonNewObject( trans_id );
1234                 osrfAppRespondComplete( ctx, ret );
1235                 jsonObjectFree(ret);
1236                 clearXactId( ctx );
1237         }
1238         return 0;
1239 }
1240
1241 /**
1242         @brief Implement the transaction.rollback method.
1243         @param ctx Pointer to the method context.
1244         @return Zero if successful, or -1 if not.
1245
1246         Issue a ROLLBACK to the database server.
1247
1248         Method parameters:
1249         - authkey (PCRUD only)
1250
1251         Return to client: Transaction ID
1252 */
1253 int rollbackTransaction ( osrfMethodContext* ctx ) {
1254         if(osrfMethodVerifyContext( ctx )) {
1255                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
1256                 return -1;
1257         }
1258
1259         if( enforce_pcrud ) {
1260                 timeout_needs_resetting = 1;
1261                 const jsonObject* user = verifyUserPCRUD( ctx );
1262                 if (!user)
1263                         return -1;
1264         }
1265
1266         // Verify that a transaction is pending
1267         const char* trans_id = getXactId( ctx );
1268         if( NULL == trans_id ) {
1269                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1270                                 "osrfMethodException", ctx->request, "No active transaction to roll back" );
1271                 return -1;
1272         }
1273
1274         dbi_result result = dbi_conn_query(writehandle, "ROLLBACK;");
1275         if (!result) {
1276                 osrfLogError(OSRF_LOG_MARK, "%s: Error rolling back transaction", modulename );
1277                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1278                                 "osrfMethodException", ctx->request, "Error rolling back transaction" );
1279                 return -1;
1280         } else {
1281                 jsonObject* ret = jsonNewObject( trans_id );
1282                 osrfAppRespondComplete( ctx, ret );
1283                 jsonObjectFree(ret);
1284                 clearXactId( ctx );
1285         }
1286         return 0;
1287 }
1288
1289 /**
1290         @brief Implement the class-specific methods.
1291         @param ctx Pointer to the method context.
1292         @return Zero if successful, or -1 if not.
1293
1294         Branch on the method type: create, retrieve, update, delete, search, or id_list.
1295
1296         The method parameters and the type of value returned to the client depend on the method
1297         type.  However, for PCRUD methods, the first method parameter should always be an
1298         authkey.
1299 */
1300 int dispatchCRUDMethod ( osrfMethodContext* ctx ) {
1301
1302         // Get the method type, then can branch on it
1303         osrfHash* method_meta = (osrfHash*) ctx->method->userData;
1304         const char* methodtype = osrfHashGet( method_meta, "methodtype" );
1305
1306         if( !strcmp( methodtype, "create" ))
1307                 return doCreate( ctx );
1308         else if( !strcmp(methodtype, "retrieve" ))
1309                 return doRetrieve( ctx );
1310         else if( !strcmp(methodtype, "update" ))
1311                 return doUpdate( ctx );
1312         else if( !strcmp(methodtype, "delete" ))
1313                 return doDelete( ctx );
1314         else if( !strcmp(methodtype, "search" ))
1315                 return doSearch( ctx );
1316         else if( !strcmp(methodtype, "id_list" ))
1317                 return doIdList( ctx );
1318         else {
1319                 osrfAppRespondComplete( ctx, NULL );      // should be unreachable...
1320                 return 0;
1321         }
1322 }
1323
1324 /**
1325         @brief Implement the "search" method.
1326         @param ctx Pointer to the method context.
1327         @return Zero if successful, or -1 if not.
1328
1329         Method parameters:
1330         - authkey (PCRUD only)
1331         - WHERE clause, as jsonObject
1332         - Other SQL clause(s), as a JSON_HASH: joins, SELECT list, LIMIT, etc.
1333
1334         Return to client: rows of the specified class that satisfy a specified WHERE clause.
1335         Optionally flesh linked fields.
1336 */
1337 static int doSearch( osrfMethodContext* ctx ) {
1338         if(osrfMethodVerifyContext( ctx )) {
1339                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
1340                 return -1;
1341         }
1342
1343         if( enforce_pcrud )
1344                 timeout_needs_resetting = 1;
1345
1346         jsonObject* where_clause;
1347         jsonObject* rest_of_query;
1348
1349         if( enforce_pcrud ) {
1350                 where_clause  = jsonObjectGetIndex( ctx->params, 1 );
1351                 rest_of_query = jsonObjectGetIndex( ctx->params, 2 );
1352         } else {
1353                 where_clause  = jsonObjectGetIndex( ctx->params, 0 );
1354                 rest_of_query = jsonObjectGetIndex( ctx->params, 1 );
1355         }
1356
1357         // Get the class metadata
1358         osrfHash* method_meta = (osrfHash*) ctx->method->userData;
1359         osrfHash* class_meta = osrfHashGet( method_meta, "class" );
1360
1361         // Do the query
1362         int err = 0;
1363         jsonObject* obj = doFieldmapperSearch( ctx, class_meta, where_clause, rest_of_query, &err );
1364         if( err ) {
1365                 osrfAppRespondComplete( ctx, NULL );
1366                 return -1;
1367         }
1368
1369         // Return each row to the client (except that some may be suppressed by PCRUD)
1370         jsonObject* cur = 0;
1371         unsigned long res_idx = 0;
1372         while((cur = jsonObjectGetIndex( obj, res_idx++ ) )) {
1373                 if( enforce_pcrud && !verifyObjectPCRUD( ctx, cur ))
1374                         continue;
1375                 osrfAppRespond( ctx, cur );
1376         }
1377         jsonObjectFree( obj );
1378
1379         osrfAppRespondComplete( ctx, NULL );
1380         return 0;
1381 }
1382
1383 /**
1384         @brief Implement the "id_list" method.
1385         @param ctx Pointer to the method context.
1386         @param err Pointer through which to return an error code.
1387         @return Zero if successful, or -1 if not.
1388
1389         Method parameters:
1390         - authkey (PCRUD only)
1391         - WHERE clause, as jsonObject
1392         - Other SQL clause(s), as a JSON_HASH: joins, LIMIT, etc.
1393
1394         Return to client: The primary key values for all rows of the relevant class that
1395         satisfy a specified WHERE clause.
1396
1397         This method relies on the assumption that every class has a primary key consisting of
1398         a single column.
1399 */
1400 static int doIdList( osrfMethodContext* ctx ) {
1401         if(osrfMethodVerifyContext( ctx )) {
1402                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
1403                 return -1;
1404         }
1405
1406         if( enforce_pcrud )
1407                 timeout_needs_resetting = 1;
1408
1409         jsonObject* where_clause;
1410         jsonObject* rest_of_query;
1411
1412         // We use the where clause without change.  But we need to massage the rest of the
1413         // query, so we work with a copy of it instead of modifying the original.
1414
1415         if( enforce_pcrud ) {
1416                 where_clause  = jsonObjectGetIndex( ctx->params, 1 );
1417                 rest_of_query = jsonObjectClone( jsonObjectGetIndex( ctx->params, 2 ) );
1418         } else {
1419                 where_clause  = jsonObjectGetIndex( ctx->params, 0 );
1420                 rest_of_query = jsonObjectClone( jsonObjectGetIndex( ctx->params, 1 ) );
1421         }
1422
1423         // Eliminate certain SQL clauses, if present.
1424         if ( rest_of_query ) {
1425                 jsonObjectRemoveKey( rest_of_query, "select" );
1426                 jsonObjectRemoveKey( rest_of_query, "no_i18n" );
1427                 jsonObjectRemoveKey( rest_of_query, "flesh" );
1428                 jsonObjectRemoveKey( rest_of_query, "flesh_columns" );
1429         } else {
1430                 rest_of_query = jsonNewObjectType( JSON_HASH );
1431         }
1432
1433         jsonObjectSetKey( rest_of_query, "no_i18n", jsonNewBoolObject( 1 ) );
1434
1435         // Get the class metadata
1436         osrfHash* method_meta = (osrfHash*) ctx->method->userData;
1437         osrfHash* class_meta = osrfHashGet( method_meta, "class" );
1438
1439         // Build a SELECT list containing just the primary key,
1440         // i.e. like { "classname":["keyname"] }
1441         jsonObject* col_list_obj = jsonNewObjectType( JSON_ARRAY );
1442
1443         // Load array with name of primary key
1444         jsonObjectPush( col_list_obj, jsonNewObject( osrfHashGet( class_meta, "primarykey" ) ) );
1445         jsonObject* select_clause = jsonNewObjectType( JSON_HASH );
1446         jsonObjectSetKey( select_clause, osrfHashGet( class_meta, "classname" ), col_list_obj );
1447
1448         jsonObjectSetKey( rest_of_query, "select", select_clause );
1449
1450         // Do the query
1451         int err = 0;
1452         jsonObject* obj =
1453                 doFieldmapperSearch( ctx, class_meta, where_clause, rest_of_query, &err );
1454
1455         jsonObjectFree( rest_of_query );
1456         if( err ) {
1457                 osrfAppRespondComplete( ctx, NULL );
1458                 return -1;
1459         }
1460
1461         // Return each primary key value to the client
1462         jsonObject* cur;
1463         unsigned long res_idx = 0;
1464         while((cur = jsonObjectGetIndex( obj, res_idx++ ) )) {
1465                 if( enforce_pcrud && !verifyObjectPCRUD( ctx, cur ))
1466                         continue;        // Suppress due to lack of permission
1467                 else
1468                         osrfAppRespond( ctx,
1469                                 oilsFMGetObject( cur, osrfHashGet( class_meta, "primarykey" ) ) );
1470         }
1471
1472         jsonObjectFree( obj );
1473         osrfAppRespondComplete( ctx, NULL );
1474         return 0;
1475 }
1476
1477 /**
1478         @brief Verify that we have a valid class reference.
1479         @param ctx Pointer to the method context.
1480         @param param Pointer to the method parameters.
1481         @return 1 if the class reference is valid, or zero if it isn't.
1482
1483         The class of the method params must match the class to which the method id devoted.
1484         For PCRUD there are additional restrictions.
1485 */
1486 static int verifyObjectClass ( osrfMethodContext* ctx, const jsonObject* param ) {
1487
1488         osrfHash* method_meta = (osrfHash*) ctx->method->userData;
1489         osrfHash* class = osrfHashGet( method_meta, "class" );
1490
1491         // Compare the method's class to the parameters' class
1492         if (!param->classname || (strcmp( osrfHashGet(class, "classname"), param->classname ))) {
1493
1494                 // Oops -- they don't match.  Complain.
1495                 growing_buffer* msg = buffer_init(128);
1496                 buffer_fadd(
1497                         msg,
1498                         "%s: %s method for type %s was passed a %s",
1499                         modulename,
1500                         osrfHashGet(method_meta, "methodtype"),
1501                         osrfHashGet(class, "classname"),
1502                         param->classname ? param->classname : "(null)"
1503                 );
1504
1505                 char* m = buffer_release(msg);
1506                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_BADREQUEST, "osrfMethodException",
1507                                 ctx->request, m );
1508                 free(m);
1509
1510                 return 0;
1511         }
1512
1513         if( enforce_pcrud )
1514                 return verifyObjectPCRUD( ctx, param );
1515         else
1516                 return 1;
1517 }
1518
1519 /**
1520         @brief (PCRUD only) Verify that the user is properly logged in.
1521         @param ctx Pointer to the method context.
1522         @return If the user is logged in, a pointer to the user object from the authentication
1523         server; otherwise NULL.
1524 */
1525 static const jsonObject* verifyUserPCRUD( osrfMethodContext* ctx ) {
1526
1527         // Get the authkey (the first method parameter)
1528         const char* auth = jsonObjectGetString( jsonObjectGetIndex( ctx->params, 0 ) );
1529
1530         // See if we have the same authkey, and a user object,
1531         // locally cached from a previous call
1532         const char* cached_authkey = getAuthkey( ctx );
1533         if( cached_authkey && !strcmp( cached_authkey, auth ) ) {
1534                 const jsonObject* cached_user = getUserLogin( ctx );
1535                 if( cached_user )
1536                         return cached_user;
1537         }
1538
1539         // We have no matching authentication data in the cache.  Authenticate from scratch.
1540         jsonObject* auth_object = jsonNewObject(auth);
1541
1542         // Fetch the user object from the authentication server
1543         jsonObject* user = oilsUtilsQuickReq( "open-ils.auth", "open-ils.auth.session.retrieve",
1544                         auth_object );
1545         jsonObjectFree(auth_object);
1546
1547         if (!user->classname || strcmp(user->classname, "au")) {
1548
1549                 growing_buffer* msg = buffer_init(128);
1550                 buffer_fadd(
1551                         msg,
1552                         "%s: permacrud received a bad auth token: %s",
1553                         modulename,
1554                         auth
1555                 );
1556
1557                 char* m = buffer_release(msg);
1558                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_UNAUTHORIZED, "osrfMethodException",
1559                                 ctx->request, m );
1560
1561                 free(m);
1562                 jsonObjectFree(user);
1563                 user = NULL;
1564         }
1565
1566         setUserLogin( ctx, user );
1567         setAuthkey( ctx, auth );
1568
1569         // Allow ourselves up to a second before we have to reset the login timeout.
1570         // It would be nice to use some fraction of the timeout interval enforced by the
1571         // authentication server, but that value is not readily available at this point.
1572         // Instead, we use a conservative default interval.
1573         time_next_reset = time( NULL ) + 1;
1574
1575         return user;
1576 }
1577
1578 static int verifyObjectPCRUD (  osrfMethodContext* ctx, const jsonObject* obj ) {
1579
1580         dbhandle = writehandle;
1581
1582         osrfHash* method_metadata = (osrfHash*) ctx->method->userData;
1583         osrfHash* class = osrfHashGet( method_metadata, "class" );
1584         const char* method_type = osrfHashGet( method_metadata, "methodtype" );
1585         int fetch = 0;
1586
1587         if ( ( *method_type == 's' || *method_type == 'i' ) ) {
1588                 method_type = "retrieve"; // search and id_list are equivalent to retrieve for this
1589         } else if ( *method_type == 'u' || *method_type == 'd' ) {
1590                 fetch = 1; // MUST go to the db for the object for update and delete
1591         }
1592
1593         osrfHash* pcrud = osrfHashGet( osrfHashGet(class, "permacrud"), method_type );
1594
1595         if (!pcrud) {
1596                 // No permacrud for this method type on this class
1597
1598                 growing_buffer* msg = buffer_init(128);
1599                 buffer_fadd(
1600                         msg,
1601                         "%s: %s on class %s has no permacrud IDL entry",
1602                         modulename,
1603                         osrfHashGet(method_metadata, "methodtype"),
1604                         osrfHashGet(class, "classname")
1605                 );
1606
1607                 char* m = buffer_release(msg);
1608                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_FORBIDDEN,
1609                                 "osrfMethodException", ctx->request, m );
1610
1611                 free(m);
1612
1613                 return 0;
1614         }
1615
1616         const jsonObject* user = verifyUserPCRUD( ctx );
1617         if (!user)
1618                 return 0;
1619
1620         int userid = atoi( oilsFMGetString( user, "id" ) );
1621
1622         osrfStringArray* permission = osrfHashGet(pcrud, "permission");
1623         osrfStringArray* local_context = osrfHashGet(pcrud, "local_context");
1624         osrfHash* foreign_context = osrfHashGet(pcrud, "foreign_context");
1625
1626         osrfStringArray* context_org_array = osrfNewStringArray(1);
1627
1628         int err = 0;
1629         char* pkey_value = NULL;
1630         if ( str_is_true( osrfHashGet(pcrud, "global_required") ) ) {
1631                 osrfLogDebug( OSRF_LOG_MARK,
1632                                 "global-level permissions required, fetching top of the org tree" );
1633
1634                 // check for perm at top of org tree
1635                 char* org_tree_root_id = org_tree_root( ctx );
1636                 if( org_tree_root_id ) {
1637                         osrfStringArrayAdd( context_org_array, org_tree_root_id );
1638                         osrfLogDebug( OSRF_LOG_MARK, "top of the org tree is %s", org_tree_root_id );
1639                 } else  {
1640                         osrfStringArrayFree( context_org_array );
1641                         return 0;
1642                 }
1643
1644         } else {
1645             osrfLogDebug( OSRF_LOG_MARK, "global-level permissions not required, "
1646                                 "fetching context org ids" );
1647             const char* pkey = osrfHashGet(class, "primarykey");
1648                 jsonObject *param = NULL;
1649
1650                 if (obj->classname) {
1651                         pkey_value = oilsFMGetString( obj, pkey );
1652                         if (!fetch)
1653                                 param = jsonObjectClone(obj);
1654                         osrfLogDebug( OSRF_LOG_MARK, "Object supplied, using primary key value of %s",
1655                                         pkey_value );
1656                 } else {
1657                         pkey_value = jsonObjectToSimpleString( obj );
1658                         fetch = 1;
1659                         osrfLogDebug( OSRF_LOG_MARK, "Object not supplied, using primary key value "
1660                                         "of %s and retrieving from the database", pkey_value );
1661                 }
1662
1663                 if (fetch) {
1664                         jsonObject* _tmp_params = single_hash( pkey, pkey_value );
1665                         jsonObject* _list = doFieldmapperSearch( ctx, class, _tmp_params, NULL, &err );
1666                         jsonObjectFree(_tmp_params);
1667
1668                         param = jsonObjectExtractIndex(_list, 0);
1669                         jsonObjectFree(_list);
1670                 }
1671
1672                 if (!param) {
1673                         osrfLogDebug( OSRF_LOG_MARK,
1674                                         "Object not found in the database with primary key %s of %s",
1675                                         pkey, pkey_value );
1676
1677                         growing_buffer* msg = buffer_init(128);
1678                         buffer_fadd(
1679                                 msg,
1680                                 "%s: no object found with primary key %s of %s",
1681                                 modulename,
1682                                 pkey,
1683                                 pkey_value
1684                         );
1685
1686                         char* m = buffer_release(msg);
1687                         osrfAppSessionStatus(
1688                                 ctx->session,
1689                                 OSRF_STATUS_INTERNALSERVERERROR,
1690                                 "osrfMethodException",
1691                                 ctx->request,
1692                                 m
1693                         );
1694
1695                         free(m);
1696                         if (pkey_value) free(pkey_value);
1697
1698                         return 0;
1699                 }
1700
1701                 if (local_context->size > 0) {
1702                         osrfLogDebug( OSRF_LOG_MARK, "%d class-local context field(s) specified",
1703                                         local_context->size);
1704                         int i = 0;
1705                         const char* lcontext = NULL;
1706                         while ( (lcontext = osrfStringArrayGetString(local_context, i++)) ) {
1707                                 osrfStringArrayAdd( context_org_array, oilsFMGetString( param, lcontext ) );
1708                                 osrfLogDebug(
1709                                         OSRF_LOG_MARK,
1710                                         "adding class-local field %s (value: %s) to the context org list",
1711                                         lcontext,
1712                                         osrfStringArrayGetString(context_org_array, context_org_array->size - 1)
1713                                 );
1714                         }
1715                 }
1716
1717                 if (foreign_context) {
1718                         unsigned long class_count = osrfHashGetCount( foreign_context );
1719                         osrfLogDebug( OSRF_LOG_MARK, "%d foreign context classes(s) specified", class_count);
1720
1721                         if (class_count > 0) {
1722
1723                                 osrfHash* fcontext = NULL;
1724                                 osrfHashIterator* class_itr = osrfNewHashIterator( foreign_context );
1725                                 while( (fcontext = osrfHashIteratorNext( class_itr ) ) ) {
1726                                         const char* class_name = osrfHashIteratorKey( class_itr );
1727                                         osrfHash* fcontext = osrfHashGet(foreign_context, class_name);
1728
1729                                         osrfLogDebug(
1730                                                 OSRF_LOG_MARK,
1731                                                 "%d foreign context fields(s) specified for class %s",
1732                                                 ((osrfStringArray*)osrfHashGet(fcontext,"context"))->size,
1733                                                 class_name
1734                                         );
1735
1736                                         char* foreign_pkey = osrfHashGet(fcontext, "field");
1737                                         char* foreign_pkey_value =
1738                                                         oilsFMGetString(param, osrfHashGet(fcontext, "fkey"));
1739
1740                                         jsonObject* _tmp_params = single_hash( foreign_pkey, foreign_pkey_value );
1741
1742                                         jsonObject* _list = doFieldmapperSearch(
1743                                                 ctx, osrfHashGet( oilsIDL(), class_name ), _tmp_params, NULL, &err );
1744
1745                                         jsonObject* _fparam = jsonObjectClone(jsonObjectGetIndex(_list, 0));
1746                                         jsonObjectFree(_tmp_params);
1747                                         jsonObjectFree(_list);
1748
1749                                         osrfStringArray* jump_list = osrfHashGet(fcontext, "jump");
1750
1751                                         if (_fparam && jump_list) {
1752                                                 const char* flink = NULL;
1753                                                 int k = 0;
1754                                                 while ( (flink = osrfStringArrayGetString(jump_list, k++)) && _fparam ) {
1755                                                         free(foreign_pkey_value);
1756
1757                                                         osrfHash* foreign_link_hash =
1758                                                                         oilsIDLFindPath( "/%s/links/%s", _fparam->classname, flink );
1759
1760                                                         foreign_pkey_value = oilsFMGetString(_fparam, flink);
1761                                                         foreign_pkey = osrfHashGet( foreign_link_hash, "key" );
1762
1763                                                         _tmp_params = single_hash( foreign_pkey, foreign_pkey_value );
1764
1765                                                         _list = doFieldmapperSearch(
1766                                                                 ctx,
1767                                                                 osrfHashGet( oilsIDL(),
1768                                                                                 osrfHashGet( foreign_link_hash, "class" ) ),
1769                                                                 _tmp_params,
1770                                                                 NULL,
1771                                                                 &err
1772                                                         );
1773
1774                                                         _fparam = jsonObjectClone(jsonObjectGetIndex(_list, 0));
1775                                                         jsonObjectFree(_tmp_params);
1776                                                         jsonObjectFree(_list);
1777                                                 }
1778                                         }
1779
1780                                         if (!_fparam) {
1781
1782                                                 growing_buffer* msg = buffer_init(128);
1783                                                 buffer_fadd(
1784                                                         msg,
1785                                                         "%s: no object found with primary key %s of %s",
1786                                                         modulename,
1787                                                         foreign_pkey,
1788                                                         foreign_pkey_value
1789                                                 );
1790
1791                                                 char* m = buffer_release(msg);
1792                                                 osrfAppSessionStatus(
1793                                                         ctx->session,
1794                                                         OSRF_STATUS_INTERNALSERVERERROR,
1795                                                         "osrfMethodException",
1796                                                         ctx->request,
1797                                                         m
1798                                                 );
1799
1800                                                 free(m);
1801                                                 osrfHashIteratorFree(class_itr);
1802                                                 free(foreign_pkey_value);
1803                                                 jsonObjectFree(param);
1804
1805                                                 return 0;
1806                                         }
1807
1808                                         free(foreign_pkey_value);
1809
1810                                         int j = 0;
1811                                         const char* foreign_field = NULL;
1812                                         while ( (foreign_field = osrfStringArrayGetString(
1813                                                          osrfHashGet(fcontext,"context" ), j++)) ) {
1814                                                 osrfStringArrayAdd( context_org_array,
1815                                                                 oilsFMGetString( _fparam, foreign_field ) );
1816                                                 osrfLogDebug(
1817                                                         OSRF_LOG_MARK,
1818                                                         "adding foreign class %s field %s (value: %s) to the context org list",
1819                                                         class_name,
1820                                                         foreign_field,
1821                                                         osrfStringArrayGetString(
1822                                                                         context_org_array, context_org_array->size - 1)
1823                                                 );
1824                                         }
1825
1826                                         jsonObjectFree(_fparam);
1827                                 }
1828
1829                                 osrfHashIteratorFree( class_itr );
1830                         }
1831                 }
1832
1833                 jsonObjectFree(param);
1834         }
1835
1836         const char* context_org = NULL;
1837         const char* perm = NULL;
1838         int OK = 0;
1839
1840         if (permission->size == 0) {
1841             osrfLogDebug( OSRF_LOG_MARK, "No permission specified for this action, passing through" );
1842                 OK = 1;
1843         }
1844
1845         int i = 0;
1846         while ( (perm = osrfStringArrayGetString(permission, i++)) ) {
1847                 int j = 0;
1848                 while ( (context_org = osrfStringArrayGetString(context_org_array, j++)) ) {
1849                         dbi_result result;
1850
1851                         if (pkey_value) {
1852                                 osrfLogDebug(
1853                                         OSRF_LOG_MARK,
1854                                         "Checking object permission [%s] for user %d "
1855                                                         "on object %s (class %s) at org %d",
1856                                         perm,
1857                                         userid,
1858                                         pkey_value,
1859                                         osrfHashGet(class, "classname"),
1860                                         atoi(context_org)
1861                                 );
1862
1863                                 result = dbi_conn_queryf(
1864                                         writehandle,
1865                                         "SELECT permission.usr_has_object_perm(%d, '%s', '%s', '%s', %d) AS has_perm;",
1866                                         userid,
1867                                         perm,
1868                                         osrfHashGet(class, "classname"),
1869                                         pkey_value,
1870                                         atoi(context_org)
1871                                 );
1872
1873                                 if (result) {
1874                                         osrfLogDebug(
1875                                                 OSRF_LOG_MARK,
1876                                                 "Received a result for object permission [%s] "
1877                                                                 "for user %d on object %s (class %s) at org %d",
1878                                                 perm,
1879                                                 userid,
1880                                                 pkey_value,
1881                                                 osrfHashGet(class, "classname"),
1882                                                 atoi(context_org)
1883                                         );
1884
1885                                         if (dbi_result_first_row(result)) {
1886                                                 jsonObject* return_val = oilsMakeJSONFromResult( result );
1887                                                 const char* has_perm = jsonObjectGetString(
1888                                                                 jsonObjectGetKeyConst(return_val, "has_perm") );
1889
1890                                                 osrfLogDebug(
1891                                                         OSRF_LOG_MARK,
1892                                                         "Status of object permission [%s] for user %d "
1893                                                                         "on object %s (class %s) at org %d is %s",
1894                                                         perm,
1895                                                         userid,
1896                                                         pkey_value,
1897                                                         osrfHashGet(class, "classname"),
1898                                                         atoi(context_org),
1899                                                         has_perm
1900                                                 );
1901
1902                                                 if ( *has_perm == 't' ) OK = 1;
1903                                                 jsonObjectFree(return_val);
1904                                         }
1905
1906                                         dbi_result_free(result);
1907                                         if (OK)
1908                                                 break;
1909                                 }
1910                         }
1911
1912                         osrfLogDebug( OSRF_LOG_MARK,
1913                                         "Checking non-object permission [%s] for user %d at org %d",
1914                                         perm, userid, atoi(context_org) );
1915                         result = dbi_conn_queryf(
1916                                 writehandle,
1917                                 "SELECT permission.usr_has_perm(%d, '%s', %d) AS has_perm;",
1918                                 userid,
1919                                 perm,
1920                                 atoi(context_org)
1921                         );
1922
1923                         if (result) {
1924                                 osrfLogDebug( OSRF_LOG_MARK,
1925                                                 "Received a result for permission [%s] for user %d at org %d",
1926                                                 perm, userid, atoi(context_org) );
1927                                 if ( dbi_result_first_row(result) ) {
1928                                         jsonObject* return_val = oilsMakeJSONFromResult( result );
1929                                         const char* has_perm = jsonObjectGetString(
1930                                                         jsonObjectGetKeyConst(return_val, "has_perm") );
1931                                         osrfLogDebug( OSRF_LOG_MARK,
1932                                                         "Status of permission [%s] for user %d at org %d is [%s]",
1933                                                         perm, userid, atoi(context_org), has_perm );
1934                                         if ( *has_perm == 't' )
1935                                                 OK = 1;
1936                                         jsonObjectFree(return_val);
1937                                 }
1938
1939                                 dbi_result_free(result);
1940                                 if (OK) break;
1941                         }
1942
1943                 }
1944                 if (OK)
1945                         break;
1946         }
1947
1948         if (pkey_value) free(pkey_value);
1949         osrfStringArrayFree(context_org_array);
1950
1951         return OK;
1952 }
1953
1954 /**
1955         @brief Look up the root of the org_unit tree.
1956         @param ctx Pointer to the method context.
1957         @return The id of the root org unit, as a character string.
1958
1959         Query actor.org_unit where parent_ou is null, and return the id as a string.
1960
1961         This function assumes that there is only one root org unit, i.e. that we
1962         have a single tree, not a forest.
1963
1964         The calling code is responsible for freeing the returned string.
1965 */
1966 static char* org_tree_root( osrfMethodContext* ctx ) {
1967
1968         static char cached_root_id[ 32 ] = "";  // extravagantly large buffer
1969         static time_t last_lookup_time = 0;
1970         time_t current_time = time( NULL );
1971
1972         if( cached_root_id[ 0 ] && ( current_time - last_lookup_time < 3600 ) ) {
1973                 // We successfully looked this up less than an hour ago.
1974                 // It's not likely to have changed since then.
1975                 return strdup( cached_root_id );
1976         }
1977         last_lookup_time = current_time;
1978
1979         int err = 0;
1980         jsonObject* where_clause = single_hash( "parent_ou", NULL );
1981         jsonObject* result = doFieldmapperSearch(
1982                 ctx, osrfHashGet( oilsIDL(), "aou" ), where_clause, NULL, &err );
1983         jsonObjectFree( where_clause );
1984
1985         jsonObject* tree_top = jsonObjectGetIndex( result, 0 );
1986
1987         if (! tree_top) {
1988                 jsonObjectFree( result );
1989
1990                 growing_buffer* msg = buffer_init(128);
1991                 OSRF_BUFFER_ADD( msg, modulename );
1992                 OSRF_BUFFER_ADD( msg,
1993                                 ": Internal error, could not find the top of the org tree (parent_ou = NULL)" );
1994
1995                 char* m = buffer_release(msg);
1996                 osrfAppSessionStatus( ctx->session,
1997                                 OSRF_STATUS_INTERNALSERVERERROR, "osrfMethodException", ctx->request, m );
1998                 free(m);
1999
2000                 cached_root_id[ 0 ] = '\0';
2001                 return NULL;
2002         }
2003
2004         char* root_org_unit_id = oilsFMGetString( tree_top, "id" );
2005         osrfLogDebug( OSRF_LOG_MARK, "Top of the org tree is %s", root_org_unit_id );
2006
2007         jsonObjectFree( result );
2008
2009         strcpy( cached_root_id, root_org_unit_id );
2010         return root_org_unit_id;
2011 }
2012
2013 /**
2014         @brief Create a JSON_HASH with a single key/value pair.
2015         @param key The key of the key/value pair.
2016         @param value the value of the key/value pair.
2017         @return Pointer to a newly created jsonObject of type JSON_HASH.
2018
2019         The value of the key/value is either a string or (if @a value is NULL) a null.
2020 */
2021 static jsonObject* single_hash( const char* key, const char* value ) {
2022         // Sanity check
2023         if( ! key ) key = "";
2024
2025         jsonObject* hash = jsonNewObjectType( JSON_HASH );
2026         jsonObjectSetKey( hash, key, jsonNewObject( value ) );
2027         return hash;
2028 }
2029
2030
2031 static int doCreate( osrfMethodContext* ctx ) {
2032         if(osrfMethodVerifyContext( ctx )) {
2033                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
2034                 return -1;
2035         }
2036
2037         if( enforce_pcrud )
2038                 timeout_needs_resetting = 1;
2039
2040         osrfHash* meta = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );
2041         jsonObject* target = NULL;
2042         jsonObject* options = NULL;
2043
2044         if( enforce_pcrud ) {
2045                 target = jsonObjectGetIndex( ctx->params, 1 );
2046                 options = jsonObjectGetIndex( ctx->params, 2 );
2047         } else {
2048                 target = jsonObjectGetIndex( ctx->params, 0 );
2049                 options = jsonObjectGetIndex( ctx->params, 1 );
2050         }
2051
2052         if ( !verifyObjectClass( ctx, target )) {
2053                 osrfAppRespondComplete( ctx, NULL );
2054                 return -1;
2055         }
2056
2057         osrfLogDebug( OSRF_LOG_MARK, "Object seems to be of the correct type" );
2058
2059         const char* trans_id = getXactId( ctx );
2060         if ( !trans_id ) {
2061                 osrfLogError( OSRF_LOG_MARK, "No active transaction -- required for CREATE" );
2062
2063                 osrfAppSessionStatus(
2064                         ctx->session,
2065                         OSRF_STATUS_BADREQUEST,
2066                         "osrfMethodException",
2067                         ctx->request,
2068                         "No active transaction -- required for CREATE"
2069                 );
2070                 osrfAppRespondComplete( ctx, NULL );
2071                 return -1;
2072         }
2073
2074         // The following test is harmless but redundant.  If a class is
2075         // readonly, we don't register a create method for it.
2076         if( str_is_true( osrfHashGet( meta, "readonly" ) ) ) {
2077                 osrfAppSessionStatus(
2078                         ctx->session,
2079                         OSRF_STATUS_BADREQUEST,
2080                         "osrfMethodException",
2081                         ctx->request,
2082                         "Cannot INSERT readonly class"
2083                 );
2084                 osrfAppRespondComplete( ctx, NULL );
2085                 return -1;
2086         }
2087
2088         // Set the last_xact_id
2089         int index = oilsIDL_ntop( target->classname, "last_xact_id" );
2090         if (index > -1) {
2091                 osrfLogDebug(OSRF_LOG_MARK, "Setting last_xact_id to %s on %s at position %d",
2092                         trans_id, target->classname, index);
2093                 jsonObjectSetIndex(target, index, jsonNewObject(trans_id));
2094         }
2095
2096         osrfLogDebug( OSRF_LOG_MARK, "There is a transaction running..." );
2097
2098         dbhandle = writehandle;
2099
2100         osrfHash* fields = osrfHashGet(meta, "fields");
2101         char* pkey = osrfHashGet(meta, "primarykey");
2102         char* seq = osrfHashGet(meta, "sequence");
2103
2104         growing_buffer* table_buf = buffer_init(128);
2105         growing_buffer* col_buf = buffer_init(128);
2106         growing_buffer* val_buf = buffer_init(128);
2107
2108         OSRF_BUFFER_ADD(table_buf, "INSERT INTO ");
2109         OSRF_BUFFER_ADD(table_buf, osrfHashGet(meta, "tablename"));
2110         OSRF_BUFFER_ADD_CHAR( col_buf, '(' );
2111         buffer_add(val_buf,"VALUES (");
2112
2113
2114         int first = 1;
2115         osrfHash* field = NULL;
2116         osrfHashIterator* field_itr = osrfNewHashIterator( fields );
2117         while( (field = osrfHashIteratorNext( field_itr ) ) ) {
2118
2119                 const char* field_name = osrfHashIteratorKey( field_itr );
2120
2121                 if( str_is_true( osrfHashGet( field, "virtual" ) ) )
2122                         continue;
2123
2124                 const jsonObject* field_object = oilsFMGetObject( target, field_name );
2125
2126                 char* value;
2127                 if (field_object && field_object->classname) {
2128                         value = oilsFMGetString(
2129                                 field_object,
2130                                 (char*)oilsIDLFindPath("/%s/primarykey", field_object->classname)
2131                         );
2132                 } else if( field_object && JSON_BOOL == field_object->type ) {
2133                         if( jsonBoolIsTrue( field_object ) )
2134                                 value = strdup( "t" );
2135                         else
2136                                 value = strdup( "f" );
2137                 } else {
2138                         value = jsonObjectToSimpleString( field_object );
2139                 }
2140
2141                 if (first) {
2142                         first = 0;
2143                 } else {
2144                         OSRF_BUFFER_ADD_CHAR( col_buf, ',' );
2145                         OSRF_BUFFER_ADD_CHAR( val_buf, ',' );
2146                 }
2147
2148                 buffer_add(col_buf, field_name);
2149
2150                 if (!field_object || field_object->type == JSON_NULL) {
2151                         buffer_add( val_buf, "DEFAULT" );
2152
2153                 } else if ( !strcmp(get_primitive( field ), "number") ) {
2154                         const char* numtype = get_datatype( field );
2155                         if ( !strcmp( numtype, "INT8") ) {
2156                                 buffer_fadd( val_buf, "%lld", atoll(value) );
2157
2158                         } else if ( !strcmp( numtype, "INT") ) {
2159                                 buffer_fadd( val_buf, "%d", atoi(value) );
2160
2161                         } else if ( !strcmp( numtype, "NUMERIC") ) {
2162                                 buffer_fadd( val_buf, "%f", atof(value) );
2163                         }
2164                 } else {
2165                         if ( dbi_conn_quote_string(writehandle, &value) ) {
2166                                 OSRF_BUFFER_ADD( val_buf, value );
2167
2168                         } else {
2169                                 osrfLogError(OSRF_LOG_MARK, "%s: Error quoting string [%s]", modulename, value);
2170                                 osrfAppSessionStatus(
2171                                         ctx->session,
2172                                         OSRF_STATUS_INTERNALSERVERERROR,
2173                                         "osrfMethodException",
2174                                         ctx->request,
2175                                         "Error quoting string -- please see the error log for more details"
2176                                 );
2177                                 free(value);
2178                                 buffer_free(table_buf);
2179                                 buffer_free(col_buf);
2180                                 buffer_free(val_buf);
2181                                 osrfAppRespondComplete( ctx, NULL );
2182                                 return -1;
2183                         }
2184                 }
2185
2186                 free(value);
2187
2188         }
2189
2190         osrfHashIteratorFree( field_itr );
2191
2192         OSRF_BUFFER_ADD_CHAR( col_buf, ')' );
2193         OSRF_BUFFER_ADD_CHAR( val_buf, ')' );
2194
2195         char* table_str = buffer_release(table_buf);
2196         char* col_str   = buffer_release(col_buf);
2197         char* val_str   = buffer_release(val_buf);
2198         growing_buffer* sql = buffer_init(128);
2199         buffer_fadd( sql, "%s %s %s;", table_str, col_str, val_str );
2200         free(table_str);
2201         free(col_str);
2202         free(val_str);
2203
2204         char* query = buffer_release(sql);
2205
2206         osrfLogDebug(OSRF_LOG_MARK, "%s: Insert SQL [%s]", modulename, query);
2207
2208         jsonObject* obj = NULL;
2209         int rc = 0;
2210
2211         dbi_result result = dbi_conn_query( writehandle, query );
2212         if (!result) {
2213                 obj = jsonNewObject(NULL);
2214                 osrfLogError(
2215                         OSRF_LOG_MARK,
2216                         "%s ERROR inserting %s object using query [%s]",
2217                         modulename,
2218                         osrfHashGet(meta, "fieldmapper"),
2219                         query
2220                 );
2221                 osrfAppSessionStatus(
2222                         ctx->session,
2223                         OSRF_STATUS_INTERNALSERVERERROR,
2224                         "osrfMethodException",
2225                         ctx->request,
2226                         "INSERT error -- please see the error log for more details"
2227                 );
2228                 rc = -1;
2229         } else {
2230
2231                 char* id = oilsFMGetString(target, pkey);
2232                 if (!id) {
2233                         unsigned long long new_id = dbi_conn_sequence_last(writehandle, seq);
2234                         growing_buffer* _id = buffer_init(10);
2235                         buffer_fadd(_id, "%lld", new_id);
2236                         id = buffer_release(_id);
2237                 }
2238
2239                 // Find quietness specification, if present
2240                 const char* quiet_str = NULL;
2241                 if ( options ) {
2242                         const jsonObject* quiet_obj = jsonObjectGetKeyConst( options, "quiet" );
2243                         if( quiet_obj )
2244                                 quiet_str = jsonObjectGetString( quiet_obj );
2245                 }
2246
2247                 if( str_is_true( quiet_str ) ) {  // if quietness is specified
2248                         obj = jsonNewObject(id);
2249                 }
2250                 else {
2251
2252                         // Fetch the row that we just inserted, so that we can return it to the client
2253                         jsonObject* where_clause = jsonNewObjectType( JSON_HASH );
2254                         jsonObjectSetKey( where_clause, pkey, jsonNewObject(id) );
2255
2256                         int err = 0;
2257                         jsonObject* list = doFieldmapperSearch( ctx, meta, where_clause, NULL, &err );
2258                         if( err )
2259                                 rc = -1;
2260                         else
2261                                 obj = jsonObjectClone( jsonObjectGetIndex( list, 0 ));
2262
2263                         jsonObjectFree( list );
2264                         jsonObjectFree( where_clause );
2265                 }
2266
2267                 free(id);
2268         }
2269
2270         free(query);
2271         osrfAppRespondComplete( ctx, obj );
2272         jsonObjectFree( obj );
2273         return rc;
2274 }
2275
2276 /**
2277         @brief Implement the retrieve method.
2278         @param ctx Pointer to the method context.
2279         @param err Pointer through which to return an error code.
2280         @return If successful, a pointer to the result to be returned to the client;
2281         otherwise NULL.
2282
2283         From the method's class, fetch a row with a specified value in the primary key.  This
2284         method relies on the database design convention that a primary key consists of a single
2285         column.
2286
2287         Method parameters:
2288         - authkey (PCRUD only)
2289         - value of the primary key for the desired row, for building the WHERE clause
2290         - a JSON_HASH containing any other SQL clauses: select, join, etc.
2291
2292         Return to client: One row from the query.
2293 */
2294 static int doRetrieve(osrfMethodContext* ctx ) {
2295         if(osrfMethodVerifyContext( ctx )) {
2296                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
2297                 return -1;
2298         }
2299
2300         if( enforce_pcrud )
2301                 timeout_needs_resetting = 1;
2302
2303         int id_pos = 0;
2304         int order_pos = 1;
2305
2306         if( enforce_pcrud ) {
2307                 id_pos = 1;
2308                 order_pos = 2;
2309         }
2310
2311         // Get the class metadata
2312         osrfHash* class_def = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );
2313
2314         // Get the value of the primary key, from a method parameter
2315         const jsonObject* id_obj = jsonObjectGetIndex(ctx->params, id_pos);
2316
2317         osrfLogDebug(
2318                 OSRF_LOG_MARK,
2319                 "%s retrieving %s object with primary key value of %s",
2320                 modulename,
2321                 osrfHashGet( class_def, "fieldmapper" ),
2322                 jsonObjectGetString( id_obj )
2323         );
2324
2325         // Build a WHERE clause based on the key value
2326         jsonObject* where_clause = jsonNewObjectType( JSON_HASH );
2327         jsonObjectSetKey(
2328                 where_clause,
2329                 osrfHashGet( class_def, "primarykey" ),  // name of key column
2330                 jsonObjectClone( id_obj )                // value of key column
2331         );
2332
2333         jsonObject* rest_of_query = jsonObjectGetIndex(ctx->params, order_pos);
2334
2335         // Do the query
2336         int err = 0;
2337         jsonObject* list = doFieldmapperSearch( ctx, class_def, where_clause, rest_of_query, &err );
2338
2339         jsonObjectFree( where_clause );
2340         if( err ) {
2341                 osrfAppRespondComplete( ctx, NULL );
2342                 return -1;
2343         }
2344
2345         jsonObject* obj = jsonObjectExtractIndex( list, 0 );
2346         jsonObjectFree( list );
2347
2348         if( enforce_pcrud ) {
2349                 if(!verifyObjectPCRUD(ctx, obj)) {
2350                         jsonObjectFree(obj);
2351
2352                         growing_buffer* msg = buffer_init(128);
2353                         OSRF_BUFFER_ADD( msg, modulename );
2354                         OSRF_BUFFER_ADD( msg, ": Insufficient permissions to retrieve object" );
2355
2356                         char* m = buffer_release(msg);
2357                         osrfAppSessionStatus( ctx->session, OSRF_STATUS_NOTALLOWED, "osrfMethodException",
2358                                         ctx->request, m );
2359                         free(m);
2360
2361                         osrfAppRespondComplete( ctx, NULL );
2362                         return -1;
2363                 }
2364         }
2365
2366         osrfAppRespondComplete( ctx, obj );
2367         jsonObjectFree( obj );
2368         return 0;
2369 }
2370
2371 static char* jsonNumberToDBString ( osrfHash* field, const jsonObject* value ) {
2372         growing_buffer* val_buf = buffer_init(32);
2373         const char* numtype = get_datatype( field );
2374
2375         if ( !strncmp( numtype, "INT", 3 ) ) {
2376                 if (value->type == JSON_NUMBER)
2377                         //buffer_fadd( val_buf, "%ld", (long)jsonObjectGetNumber(value) );
2378                         buffer_fadd( val_buf, jsonObjectGetString( value ) );
2379                 else {
2380                         buffer_fadd( val_buf, jsonObjectGetString( value ) );
2381                 }
2382
2383         } else if ( !strcmp( numtype, "NUMERIC" ) ) {
2384                 if (value->type == JSON_NUMBER)
2385                         buffer_fadd( val_buf, jsonObjectGetString( value ) );
2386                 else {
2387                         buffer_fadd( val_buf, jsonObjectGetString( value ) );
2388                 }
2389
2390         } else {
2391                 // Presumably this was really intended ot be a string, so quote it
2392                 char* str = jsonObjectToSimpleString( value );
2393                 if ( dbi_conn_quote_string(dbhandle, &str) ) {
2394                         OSRF_BUFFER_ADD( val_buf, str );
2395                         free(str);
2396                 } else {
2397                         osrfLogError(OSRF_LOG_MARK, "%s: Error quoting key string [%s]", modulename, str);
2398                         free(str);
2399                         buffer_free(val_buf);
2400                         return NULL;
2401                 }
2402         }
2403
2404         return buffer_release(val_buf);
2405 }
2406
2407 static char* searchINPredicate (const char* class_alias, osrfHash* field,
2408                 jsonObject* node, const char* op, osrfMethodContext* ctx ) {
2409         growing_buffer* sql_buf = buffer_init(32);
2410
2411         buffer_fadd(
2412                 sql_buf,
2413                 "\"%s\".%s ",
2414                 class_alias,
2415                 osrfHashGet(field, "name")
2416         );
2417
2418         if (!op) {
2419                 buffer_add(sql_buf, "IN (");
2420         } else if (!(strcasecmp(op,"not in"))) {
2421                 buffer_add(sql_buf, "NOT IN (");
2422         } else {
2423                 buffer_add(sql_buf, "IN (");
2424         }
2425
2426         if (node->type == JSON_HASH) {
2427                 // subquery predicate
2428                 char* subpred = buildQuery( ctx, node, SUBSELECT );
2429                 if( ! subpred ) {
2430                         buffer_free( sql_buf );
2431                         return NULL;
2432                 }
2433
2434                 buffer_add(sql_buf, subpred);
2435                 free(subpred);
2436
2437         } else if (node->type == JSON_ARRAY) {
2438                 // literal value list
2439                 int in_item_index = 0;
2440                 int in_item_first = 1;
2441                 const jsonObject* in_item;
2442                 while ( (in_item = jsonObjectGetIndex(node, in_item_index++)) ) {
2443
2444                         if (in_item_first)
2445                                 in_item_first = 0;
2446                         else
2447                                 buffer_add(sql_buf, ", ");
2448
2449                         // Sanity check
2450                         if ( in_item->type != JSON_STRING && in_item->type != JSON_NUMBER ) {
2451                                 osrfLogError( OSRF_LOG_MARK,
2452                                                 "%s: Expected string or number within IN list; found %s",
2453                                                 modulename, json_type( in_item->type ) );
2454                                 buffer_free(sql_buf);
2455                                 return NULL;
2456                         }
2457
2458                         // Append the literal value -- quoted if not a number
2459                         if ( JSON_NUMBER == in_item->type ) {
2460                                 char* val = jsonNumberToDBString( field, in_item );
2461                                 OSRF_BUFFER_ADD( sql_buf, val );
2462                                 free(val);
2463
2464                         } else if ( !strcmp( get_primitive( field ), "number") ) {
2465                                 char* val = jsonNumberToDBString( field, in_item );
2466                                 OSRF_BUFFER_ADD( sql_buf, val );
2467                                 free(val);
2468
2469                         } else {
2470                                 char* key_string = jsonObjectToSimpleString(in_item);
2471                                 if ( dbi_conn_quote_string(dbhandle, &key_string) ) {
2472                                         OSRF_BUFFER_ADD( sql_buf, key_string );
2473                                         free(key_string);
2474                                 } else {
2475                                         osrfLogError(OSRF_LOG_MARK,
2476                                                         "%s: Error quoting key string [%s]", modulename, key_string);
2477                                         free(key_string);
2478                                         buffer_free(sql_buf);
2479                                         return NULL;
2480                                 }
2481                         }
2482                 }
2483
2484                 if( in_item_first ) {
2485                         osrfLogError(OSRF_LOG_MARK, "%s: Empty IN list", modulename );
2486                         buffer_free( sql_buf );
2487                         return NULL;
2488                 }
2489         } else {
2490                 osrfLogError(OSRF_LOG_MARK, "%s: Expected object or array for IN clause; found %s",
2491                         modulename, json_type( node->type ) );
2492                 buffer_free(sql_buf);
2493                 return NULL;
2494         }
2495
2496         OSRF_BUFFER_ADD_CHAR( sql_buf, ')' );
2497
2498         return buffer_release(sql_buf);
2499 }
2500
2501 // Receive a JSON_ARRAY representing a function call.  The first
2502 // entry in the array is the function name.  The rest are parameters.
2503 static char* searchValueTransform( const jsonObject* array ) {
2504
2505         if( array->size < 1 ) {
2506                 osrfLogError( OSRF_LOG_MARK, "%s: Empty array for value transform", modulename );
2507                 return NULL;
2508         }
2509
2510         // Get the function name
2511         jsonObject* func_item = jsonObjectGetIndex( array, 0 );
2512         if( func_item->type != JSON_STRING ) {
2513                 osrfLogError(OSRF_LOG_MARK, "%s: Error: expected function name, found %s",
2514                         modulename, json_type( func_item->type ) );
2515                 return NULL;
2516         }
2517
2518         growing_buffer* sql_buf = buffer_init(32);
2519
2520         OSRF_BUFFER_ADD( sql_buf, jsonObjectGetString( func_item ) );
2521         OSRF_BUFFER_ADD( sql_buf, "( " );
2522
2523         // Get the parameters
2524         int func_item_index = 1;   // We already grabbed the zeroth entry
2525         while ( (func_item = jsonObjectGetIndex(array, func_item_index++)) ) {
2526
2527                 // Add a separator comma, if we need one
2528                 if( func_item_index > 2 )
2529                         buffer_add( sql_buf, ", " );
2530
2531                 // Add the current parameter
2532                 if (func_item->type == JSON_NULL) {
2533                         buffer_add( sql_buf, "NULL" );
2534                 } else {
2535                         char* val = jsonObjectToSimpleString(func_item);
2536                         if ( dbi_conn_quote_string(dbhandle, &val) ) {
2537                                 OSRF_BUFFER_ADD( sql_buf, val );
2538                                 free(val);
2539                         } else {
2540                                 osrfLogError(OSRF_LOG_MARK, "%s: Error quoting key string [%s]", modulename, val);
2541                                 buffer_free(sql_buf);
2542                                 free(val);
2543                                 return NULL;
2544                         }
2545                 }
2546         }
2547
2548         buffer_add( sql_buf, " )" );
2549
2550         return buffer_release(sql_buf);
2551 }
2552
2553 static char* searchFunctionPredicate (const char* class_alias, osrfHash* field,
2554                 const jsonObject* node, const char* op) {
2555
2556         if( ! is_good_operator( op ) ) {
2557                 osrfLogError( OSRF_LOG_MARK, "%s: Invalid operator [%s]", modulename, op );
2558                 return NULL;
2559         }
2560
2561         char* val = searchValueTransform(node);
2562         if( !val )
2563                 return NULL;
2564
2565         growing_buffer* sql_buf = buffer_init(32);
2566         buffer_fadd(
2567                 sql_buf,
2568                 "\"%s\".%s %s %s",
2569                 class_alias,
2570                 osrfHashGet(field, "name"),
2571                 op,
2572                 val
2573         );
2574
2575         free(val);
2576
2577         return buffer_release(sql_buf);
2578 }
2579
2580 // class_alias is a class name or other table alias
2581 // field is a field definition as stored in the IDL
2582 // node comes from the method parameter, and may represent an entry in the SELECT list
2583 static char* searchFieldTransform (const char* class_alias, osrfHash* field, const jsonObject* node) {
2584         growing_buffer* sql_buf = buffer_init(32);
2585
2586         const char* field_transform = jsonObjectGetString(
2587                 jsonObjectGetKeyConst( node, "transform" ) );
2588         const char* transform_subcolumn = jsonObjectGetString(
2589                 jsonObjectGetKeyConst( node, "result_field" ) );
2590
2591         if(transform_subcolumn) {
2592                 if( ! is_identifier( transform_subcolumn ) ) {
2593                         osrfLogError( OSRF_LOG_MARK, "%s: Invalid subfield name: \"%s\"\n",
2594                                         modulename, transform_subcolumn );
2595                         buffer_free( sql_buf );
2596                         return NULL;
2597                 }
2598                 OSRF_BUFFER_ADD_CHAR( sql_buf, '(' );    // enclose transform in parentheses
2599         }
2600
2601         if (field_transform) {
2602
2603                 if( ! is_identifier( field_transform ) ) {
2604                         osrfLogError( OSRF_LOG_MARK, "%s: Expected function name, found \"%s\"\n",
2605                                         modulename, field_transform );
2606                         buffer_free( sql_buf );
2607                         return NULL;
2608                 }
2609
2610                 if( obj_is_true( jsonObjectGetKeyConst( node, "distinct" ) ) ) {
2611                         buffer_fadd( sql_buf, "%s(DISTINCT \"%s\".%s",
2612                                 field_transform, class_alias, osrfHashGet(field, "name"));
2613                 } else {
2614                         buffer_fadd( sql_buf, "%s(\"%s\".%s",
2615                                 field_transform, class_alias, osrfHashGet(field, "name"));
2616                 }
2617
2618                 const jsonObject* array = jsonObjectGetKeyConst( node, "params" );
2619
2620                 if (array) {
2621                         if( array->type != JSON_ARRAY ) {
2622                                 osrfLogError( OSRF_LOG_MARK,
2623                                         "%s: Expected JSON_ARRAY for function params; found %s",
2624                                         modulename, json_type( array->type ) );
2625                                 buffer_free( sql_buf );
2626                                 return NULL;
2627                         }
2628                         int func_item_index = 0;
2629                         jsonObject* func_item;
2630                         while ( (func_item = jsonObjectGetIndex(array, func_item_index++)) ) {
2631
2632                                 char* val = jsonObjectToSimpleString(func_item);
2633
2634                                 if ( !val ) {
2635                                         buffer_add( sql_buf, ",NULL" );
2636                                 } else if ( dbi_conn_quote_string(dbhandle, &val) ) {
2637                                         OSRF_BUFFER_ADD_CHAR( sql_buf, ',' );
2638                                         OSRF_BUFFER_ADD( sql_buf, val );
2639                                 } else {
2640                                         osrfLogError( OSRF_LOG_MARK,
2641                                                         "%s: Error quoting key string [%s]", modulename, val);
2642                                         free(val);
2643                                         buffer_free(sql_buf);
2644                                         return NULL;
2645                                 }
2646                                 free(val);
2647                         }
2648                 }
2649
2650                 buffer_add( sql_buf, " )" );
2651
2652         } else {
2653                 buffer_fadd( sql_buf, "\"%s\".%s", class_alias, osrfHashGet(field, "name"));
2654         }
2655
2656         if (transform_subcolumn)
2657                 buffer_fadd( sql_buf, ").\"%s\"", transform_subcolumn );
2658
2659         return buffer_release(sql_buf);
2660 }
2661
2662 static char* searchFieldTransformPredicate( const ClassInfo* class_info, osrfHash* field,
2663                 const jsonObject* node, const char* op ) {
2664
2665         if( ! is_good_operator( op ) ) {
2666                 osrfLogError(OSRF_LOG_MARK, "%s: Error: Invalid operator %s", modulename, op);
2667                 return NULL;
2668         }
2669
2670         char* field_transform = searchFieldTransform( class_info->alias, field, node );
2671         if( ! field_transform )
2672                 return NULL;
2673         char* value = NULL;
2674         int extra_parens = 0;   // boolean
2675
2676         const jsonObject* value_obj = jsonObjectGetKeyConst( node, "value" );
2677         if ( ! value_obj ) {
2678                 value = searchWHERE( node, class_info, AND_OP_JOIN, NULL );
2679                 if( !value ) {
2680                         osrfLogError( OSRF_LOG_MARK, "%s: Error building condition for field transform",
2681                                 modulename );
2682                         free(field_transform);
2683                         return NULL;
2684                 }
2685                 extra_parens = 1;
2686         } else if ( value_obj->type == JSON_ARRAY ) {
2687                 value = searchValueTransform( value_obj );
2688                 if( !value ) {
2689                         osrfLogError( OSRF_LOG_MARK,
2690                                 "%s: Error building value transform for field transform", modulename );
2691                         free( field_transform );
2692                         return NULL;
2693                 }
2694         } else if ( value_obj->type == JSON_HASH ) {
2695                 value = searchWHERE( value_obj, class_info, AND_OP_JOIN, NULL );
2696                 if( !value ) {
2697                         osrfLogError( OSRF_LOG_MARK, "%s: Error building predicate for field transform",
2698                                 modulename );
2699                         free(field_transform);
2700                         return NULL;
2701                 }
2702                 extra_parens = 1;
2703         } else if ( value_obj->type == JSON_NUMBER ) {
2704                 value = jsonNumberToDBString( field, value_obj );
2705         } else if ( value_obj->type == JSON_NULL ) {
2706                 osrfLogError( OSRF_LOG_MARK,
2707                         "%s: Error building predicate for field transform: null value", modulename );
2708                 free(field_transform);
2709                 return NULL;
2710         } else if ( value_obj->type == JSON_BOOL ) {
2711                 osrfLogError( OSRF_LOG_MARK,
2712                         "%s: Error building predicate for field transform: boolean value", modulename );
2713                 free(field_transform);
2714                 return NULL;
2715         } else {
2716                 if ( !strcmp( get_primitive( field ), "number") ) {
2717                         value = jsonNumberToDBString( field, value_obj );
2718                 } else {
2719                         value = jsonObjectToSimpleString( value_obj );
2720                         if ( !dbi_conn_quote_string(dbhandle, &value) ) {
2721                                 osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key string [%s]",
2722                                         modulename, value );
2723                                 free(value);
2724                                 free(field_transform);
2725                                 return NULL;
2726                         }
2727                 }
2728         }
2729
2730         const char* left_parens  = "";
2731         const char* right_parens = "";
2732
2733         if( extra_parens ) {
2734                 left_parens  = "(";
2735                 right_parens = ")";
2736         }
2737
2738         growing_buffer* sql_buf = buffer_init(32);
2739
2740         buffer_fadd(
2741                 sql_buf,
2742                 "%s%s %s %s %s %s%s",
2743                 left_parens,
2744                 field_transform,
2745                 op,
2746                 left_parens,
2747                 value,
2748                 right_parens,
2749                 right_parens
2750         );
2751
2752         free(value);
2753         free(field_transform);
2754
2755         return buffer_release(sql_buf);
2756 }
2757
2758 static char* searchSimplePredicate (const char* op, const char* class_alias,
2759                 osrfHash* field, const jsonObject* node) {
2760
2761         if( ! is_good_operator( op ) ) {
2762                 osrfLogError( OSRF_LOG_MARK, "%s: Invalid operator [%s]", modulename, op );
2763                 return NULL;
2764         }
2765
2766         char* val = NULL;
2767
2768         // Get the value to which we are comparing the specified column
2769         if (node->type != JSON_NULL) {
2770                 if ( node->type == JSON_NUMBER ) {
2771                         val = jsonNumberToDBString( field, node );
2772                 } else if ( !strcmp( get_primitive( field ), "number" ) ) {
2773                         val = jsonNumberToDBString( field, node );
2774                 } else {
2775                         val = jsonObjectToSimpleString(node);
2776                 }
2777         }
2778
2779         if( val ) {
2780                 if( JSON_NUMBER != node->type && strcmp( get_primitive( field ), "number") ) {
2781                         // Value is not numeric; enclose it in quotes
2782                         if ( !dbi_conn_quote_string( dbhandle, &val ) ) {
2783                                 osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key string [%s]",
2784                                         modulename, val );
2785                                 free( val );
2786                                 return NULL;
2787                         }
2788                 }
2789         } else {
2790                 // Compare to a null value
2791                 val = strdup( "NULL" );
2792                 if (strcmp( op, "=" ))
2793                         op = "IS NOT";
2794                 else
2795                         op = "IS";
2796         }
2797
2798         growing_buffer* sql_buf = buffer_init(32);
2799         buffer_fadd( sql_buf, "\"%s\".%s %s %s", class_alias, osrfHashGet(field, "name"), op, val );
2800         char* pred = buffer_release( sql_buf );
2801
2802         free(val);
2803
2804         return pred;
2805 }
2806
2807 static char* searchBETWEENPredicate (const char* class_alias,
2808                 osrfHash* field, const jsonObject* node) {
2809
2810         const jsonObject* x_node = jsonObjectGetIndex( node, 0 );
2811         const jsonObject* y_node = jsonObjectGetIndex( node, 1 );
2812
2813         if( NULL == y_node ) {
2814                 osrfLogError( OSRF_LOG_MARK, "%s: Not enough operands for BETWEEN operator", modulename );
2815                 return NULL;
2816         }
2817         else if( NULL != jsonObjectGetIndex( node, 2 ) ) {
2818                 osrfLogError( OSRF_LOG_MARK, "%s: Too many operands for BETWEEN operator", modulename );
2819                 return NULL;
2820         }
2821
2822         char* x_string;
2823         char* y_string;
2824
2825         if ( !strcmp( get_primitive( field ), "number") ) {
2826                 x_string = jsonNumberToDBString(field, x_node);
2827                 y_string = jsonNumberToDBString(field, y_node);
2828
2829         } else {
2830                 x_string = jsonObjectToSimpleString(x_node);
2831                 y_string = jsonObjectToSimpleString(y_node);
2832                 if ( !(dbi_conn_quote_string(dbhandle, &x_string) && dbi_conn_quote_string(dbhandle, &y_string)) ) {
2833                         osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key strings [%s] and [%s]",
2834                                         modulename, x_string, y_string );
2835                         free(x_string);
2836                         free(y_string);
2837                         return NULL;
2838                 }
2839         }
2840
2841         growing_buffer* sql_buf = buffer_init(32);
2842         buffer_fadd( sql_buf, "\"%s\".%s BETWEEN %s AND %s",
2843                         class_alias, osrfHashGet(field, "name"), x_string, y_string );
2844         free(x_string);
2845         free(y_string);
2846
2847         return buffer_release(sql_buf);
2848 }
2849
2850 static char* searchPredicate ( const ClassInfo* class_info, osrfHash* field,
2851                                                            jsonObject* node, osrfMethodContext* ctx ) {
2852
2853         char* pred = NULL;
2854         if (node->type == JSON_ARRAY) { // equality IN search
2855                 pred = searchINPredicate( class_info->alias, field, node, NULL, ctx );
2856         } else if (node->type == JSON_HASH) { // other search
2857                 jsonIterator* pred_itr = jsonNewIterator( node );
2858                 if( !jsonIteratorHasNext( pred_itr ) ) {
2859                         osrfLogError( OSRF_LOG_MARK, "%s: Empty predicate for field \"%s\"",
2860                                         modulename, osrfHashGet(field, "name" ));
2861                 } else {
2862                         jsonObject* pred_node = jsonIteratorNext( pred_itr );
2863
2864                         // Verify that there are no additional predicates
2865                         if( jsonIteratorHasNext( pred_itr ) ) {
2866                                 osrfLogError( OSRF_LOG_MARK, "%s: Multiple predicates for field \"%s\"",
2867                                                 modulename, osrfHashGet(field, "name" ));
2868                         } else if ( !(strcasecmp( pred_itr->key,"between" )) )
2869                                 pred = searchBETWEENPredicate( class_info->alias, field, pred_node );
2870                         else if ( !(strcasecmp( pred_itr->key,"in" ))
2871                                         || !(strcasecmp( pred_itr->key,"not in" )) )
2872                                 pred = searchINPredicate(
2873                                         class_info->alias, field, pred_node, pred_itr->key, ctx );
2874                         else if ( pred_node->type == JSON_ARRAY )
2875                                 pred = searchFunctionPredicate(
2876                                         class_info->alias, field, pred_node, pred_itr->key );
2877                         else if ( pred_node->type == JSON_HASH )
2878                                 pred = searchFieldTransformPredicate(
2879                                         class_info, field, pred_node, pred_itr->key );
2880                         else
2881                                 pred = searchSimplePredicate( pred_itr->key, class_info->alias, field, pred_node );
2882                 }
2883                 jsonIteratorFree(pred_itr);
2884
2885         } else if (node->type == JSON_NULL) { // IS NULL search
2886                 growing_buffer* _p = buffer_init(64);
2887                 buffer_fadd(
2888                         _p,
2889                         "\"%s\".%s IS NULL",
2890                         class_info->class_name,
2891                         osrfHashGet(field, "name")
2892                 );
2893                 pred = buffer_release(_p);
2894         } else { // equality search
2895                 pred = searchSimplePredicate( "=", class_info->alias, field, node );
2896         }
2897
2898         return pred;
2899
2900 }
2901
2902
2903 /*
2904
2905 join : {
2906         acn : {
2907                 field : record,
2908                 fkey : id
2909                 type : left
2910                 filter_op : or
2911                 filter : { ... },
2912                 join : {
2913                         acp : {
2914                                 field : call_number,
2915                                 fkey : id,
2916                                 filter : { ... },
2917                         },
2918                 },
2919         },
2920         mrd : {
2921                 field : record,
2922                 type : inner
2923                 fkey : id,
2924                 filter : { ... },
2925         }
2926 }
2927
2928 */
2929
2930 static char* searchJOIN ( const jsonObject* join_hash, const ClassInfo* left_info ) {
2931
2932         const jsonObject* working_hash;
2933         jsonObject* freeable_hash = NULL;
2934
2935         if (join_hash->type == JSON_HASH) {
2936                 working_hash = join_hash;
2937         } else if (join_hash->type == JSON_STRING) {
2938                 // turn it into a JSON_HASH by creating a wrapper
2939                 // around a copy of the original
2940                 const char* _tmp = jsonObjectGetString( join_hash );
2941                 freeable_hash = jsonNewObjectType(JSON_HASH);
2942                 jsonObjectSetKey(freeable_hash, _tmp, NULL);
2943                 working_hash = freeable_hash;
2944         } else {
2945                 osrfLogError(
2946                         OSRF_LOG_MARK,
2947                         "%s: JOIN failed; expected JSON object type not found",
2948                         modulename
2949                 );
2950                 return NULL;
2951         }
2952
2953         growing_buffer* join_buf = buffer_init(128);
2954         const char* leftclass = left_info->class_name;
2955
2956         jsonObject* snode = NULL;
2957         jsonIterator* search_itr = jsonNewIterator( working_hash );
2958
2959         while ( (snode = jsonIteratorNext( search_itr )) ) {
2960                 const char* right_alias = search_itr->key;
2961                 const char* class =
2962                                 jsonObjectGetString( jsonObjectGetKeyConst( snode, "class" ) );
2963                 if( ! class )
2964                         class = right_alias;
2965
2966                 const ClassInfo* right_info = add_joined_class( right_alias, class );
2967                 if( !right_info ) {
2968                         osrfLogError(
2969                                 OSRF_LOG_MARK,
2970                                 "%s: JOIN failed.  Class \"%s\" not resolved in IDL",
2971                                 modulename,
2972                                 search_itr->key
2973                         );
2974                         jsonIteratorFree( search_itr );
2975                         buffer_free( join_buf );
2976                         if( freeable_hash )
2977                                 jsonObjectFree( freeable_hash );
2978                         return NULL;
2979                 }
2980                 osrfHash* links    = right_info->links;
2981                 const char* table  = right_info->source_def;
2982
2983                 const char* fkey  = jsonObjectGetString( jsonObjectGetKeyConst( snode, "fkey" ) );
2984                 const char* field = jsonObjectGetString( jsonObjectGetKeyConst( snode, "field" ) );
2985
2986                 if (field && !fkey) {
2987                         // Look up the corresponding join column in the IDL.
2988                         // The link must be defined in the child table,
2989                         // and point to the right parent table.
2990                         osrfHash* idl_link = (osrfHash*) osrfHashGet( links, field );
2991                         const char* reltype = NULL;
2992                         const char* other_class = NULL;
2993                         reltype = osrfHashGet( idl_link, "reltype" );
2994                         if( reltype && strcmp( reltype, "has_many" ) )
2995                                 other_class = osrfHashGet( idl_link, "class" );
2996                         if( other_class && !strcmp( other_class, leftclass ) )
2997                                 fkey = osrfHashGet( idl_link, "key" );
2998                         if (!fkey) {
2999                                 osrfLogError(
3000                                         OSRF_LOG_MARK,
3001                                         "%s: JOIN failed.  No link defined from %s.%s to %s",
3002                                         modulename,
3003                                         class,
3004                                         field,
3005                                         leftclass
3006                                 );
3007                                 buffer_free(join_buf);
3008                                 if(freeable_hash)
3009                                         jsonObjectFree(freeable_hash);
3010                                 jsonIteratorFree(search_itr);
3011                                 return NULL;
3012                         }
3013
3014                 } else if (!field && fkey) {
3015                         // Look up the corresponding join column in the IDL.
3016                         // The link must be defined in the child table,
3017                         // and point to the right parent table.
3018                         osrfHash* left_links = left_info->links;
3019                         osrfHash* idl_link = (osrfHash*) osrfHashGet( left_links, fkey );
3020                         const char* reltype = NULL;
3021                         const char* other_class = NULL;
3022                         reltype = osrfHashGet( idl_link, "reltype" );
3023                         if( reltype && strcmp( reltype, "has_many" ) )
3024                                 other_class = osrfHashGet( idl_link, "class" );
3025                         if( other_class && !strcmp( other_class, class ) )
3026                                 field = osrfHashGet( idl_link, "key" );
3027                         if (!field) {
3028                                 osrfLogError(
3029                                         OSRF_LOG_MARK,
3030                                         "%s: JOIN failed.  No link defined from %s.%s to %s",
3031                                         modulename,
3032                                         leftclass,
3033                                         fkey,
3034                                         class
3035                                 );
3036                                 buffer_free(join_buf);
3037                                 if(freeable_hash)
3038                                         jsonObjectFree(freeable_hash);
3039                                 jsonIteratorFree(search_itr);
3040                                 return NULL;
3041                         }
3042
3043                 } else if (!field && !fkey) {
3044                         osrfHash* left_links = left_info->links;
3045
3046                         // For each link defined for the left class:
3047                         // see if the link references the joined class
3048                         osrfHashIterator* itr = osrfNewHashIterator( left_links );
3049                         osrfHash* curr_link = NULL;
3050                         while( (curr_link = osrfHashIteratorNext( itr ) ) ) {
3051                                 const char* other_class = osrfHashGet( curr_link, "class" );
3052                                 if( other_class && !strcmp( other_class, class ) ) {
3053
3054                                         // In the IDL, the parent class doesn't always know then names of the child
3055                                         // columns that are pointing to it, so don't use that end of the link
3056                                         const char* reltype = osrfHashGet( curr_link, "reltype" );
3057                                         if( reltype && strcmp( reltype, "has_many" ) ) {
3058                                                 // Found a link between the classes
3059                                                 fkey = osrfHashIteratorKey( itr );
3060                                                 field = osrfHashGet( curr_link, "key" );
3061                                                 break;
3062                                         }
3063                                 }
3064                         }
3065                         osrfHashIteratorFree( itr );
3066
3067                         if (!field || !fkey) {
3068                                 // Do another such search, with the classes reversed
3069
3070                                 // For each link defined for the joined class:
3071                                 // see if the link references the left class
3072                                 osrfHashIterator* itr = osrfNewHashIterator( links );
3073                                 osrfHash* curr_link = NULL;
3074                                 while( (curr_link = osrfHashIteratorNext( itr ) ) ) {
3075                                         const char* other_class = osrfHashGet( curr_link, "class" );
3076                                         if( other_class && !strcmp( other_class, leftclass ) ) {
3077
3078                                                 // In the IDL, the parent class doesn't know then names of the child
3079                                                 // columns that are pointing to it, so don't use that end of the link
3080                                                 const char* reltype = osrfHashGet( curr_link, "reltype" );
3081                                                 if( reltype && strcmp( reltype, "has_many" ) ) {
3082                                                         // Found a link between the classes
3083                                                         field = osrfHashIteratorKey( itr );
3084                                                         fkey = osrfHashGet( curr_link, "key" );
3085                                                         break;
3086                                                 }
3087                                         }
3088                                 }
3089                                 osrfHashIteratorFree( itr );
3090                         }
3091
3092                         if (!field || !fkey) {
3093                                 osrfLogError(
3094                                         OSRF_LOG_MARK,
3095                                         "%s: JOIN failed.  No link defined between %s and %s",
3096                                         modulename,
3097                                         leftclass,
3098                                         class
3099                                 );
3100                                 buffer_free(join_buf);
3101                                 if(freeable_hash)
3102                                         jsonObjectFree(freeable_hash);
3103                                 jsonIteratorFree(search_itr);
3104                                 return NULL;
3105                         }
3106                 }
3107
3108                 const char* type = jsonObjectGetString( jsonObjectGetKeyConst( snode, "type" ) );
3109                 if (type) {
3110                         if ( !strcasecmp(type,"left") ) {
3111                                 buffer_add(join_buf, " LEFT JOIN");
3112                         } else if ( !strcasecmp(type,"right") ) {
3113                                 buffer_add(join_buf, " RIGHT JOIN");
3114                         } else if ( !strcasecmp(type,"full") ) {
3115                                 buffer_add(join_buf, " FULL JOIN");
3116                         } else {
3117                                 buffer_add(join_buf, " INNER JOIN");
3118                         }
3119                 } else {
3120                         buffer_add(join_buf, " INNER JOIN");
3121                 }
3122
3123                 buffer_fadd(join_buf, " %s AS \"%s\" ON ( \"%s\".%s = \"%s\".%s",
3124                                         table, right_alias, right_alias, field, left_info->alias, fkey);
3125
3126                 // Add any other join conditions as specified by "filter"
3127                 const jsonObject* filter = jsonObjectGetKeyConst( snode, "filter" );
3128                 if (filter) {
3129                         const char* filter_op = jsonObjectGetString(
3130                                 jsonObjectGetKeyConst( snode, "filter_op" ) );
3131                         if ( filter_op && !strcasecmp("or",filter_op) ) {
3132                                 buffer_add( join_buf, " OR " );
3133                         } else {
3134                                 buffer_add( join_buf, " AND " );
3135                         }
3136
3137                         char* jpred = searchWHERE( filter, right_info, AND_OP_JOIN, NULL );
3138                         if( jpred ) {
3139                                 OSRF_BUFFER_ADD_CHAR( join_buf, ' ' );
3140                                 OSRF_BUFFER_ADD( join_buf, jpred );
3141                                 free(jpred);
3142                         } else {
3143                                 osrfLogError(
3144                                         OSRF_LOG_MARK,
3145                                         "%s: JOIN failed.  Invalid conditional expression.",
3146                                         modulename
3147                                 );
3148                                 jsonIteratorFree( search_itr );
3149                                 buffer_free( join_buf );
3150                                 if( freeable_hash )
3151                                         jsonObjectFree( freeable_hash );
3152                                 return NULL;
3153                         }
3154                 }
3155
3156                 buffer_add(join_buf, " ) ");
3157
3158                 // Recursively add a nested join, if one is present
3159                 const jsonObject* join_filter = jsonObjectGetKeyConst( snode, "join" );
3160                 if (join_filter) {
3161                         char* jpred = searchJOIN( join_filter, right_info );
3162                         if( jpred ) {
3163                                 OSRF_BUFFER_ADD_CHAR( join_buf, ' ' );
3164                                 OSRF_BUFFER_ADD( join_buf, jpred );
3165                                 free(jpred);
3166                         } else {
3167                                 osrfLogError(  OSRF_LOG_MARK, "%s: Invalid nested join.", modulename );
3168                                 jsonIteratorFree( search_itr );
3169                                 buffer_free( join_buf );
3170                                 if( freeable_hash )
3171                                         jsonObjectFree( freeable_hash );
3172                                 return NULL;
3173                         }
3174                 }
3175         }
3176
3177         if(freeable_hash)
3178                 jsonObjectFree(freeable_hash);
3179         jsonIteratorFree(search_itr);
3180
3181         return buffer_release(join_buf);
3182 }
3183
3184 /*
3185
3186 { +class : { -or|-and : { field : { op : value }, ... } ... }, ... }
3187 { +class : { -or|-and : [ { field : { op : value }, ... }, ...] ... }, ... }
3188 [ { +class : { -or|-and : [ { field : { op : value }, ... }, ...] ... }, ... }, ... ]
3189
3190 Generate code to express a set of conditions, as for a WHERE clause.  Parameters:
3191
3192 search_hash is the JSON expression of the conditions.
3193 meta is the class definition from the IDL, for the relevant table.
3194 opjoin_type indicates whether multiple conditions, if present, should be
3195         connected by AND or OR.
3196 osrfMethodContext is loaded with all sorts of stuff, but all we do with it here is
3197         to pass it to other functions -- and all they do with it is to use the session
3198         and request members to send error messages back to the client.
3199
3200 */
3201
3202 static char* searchWHERE ( const jsonObject* search_hash, const ClassInfo* class_info,
3203                 int opjoin_type, osrfMethodContext* ctx ) {
3204
3205         osrfLogDebug(
3206                 OSRF_LOG_MARK,
3207                 "%s: Entering searchWHERE; search_hash addr = %p, meta addr = %p, "
3208                 "opjoin_type = %d, ctx addr = %p",
3209                 modulename,
3210                 search_hash,
3211                 class_info->class_def,
3212                 opjoin_type,
3213                 ctx
3214         );
3215
3216         growing_buffer* sql_buf = buffer_init(128);
3217
3218         jsonObject* node = NULL;
3219
3220         int first = 1;
3221         if ( search_hash->type == JSON_ARRAY ) {
3222                 osrfLogDebug( OSRF_LOG_MARK,
3223                   "%s: In WHERE clause, condition type is JSON_ARRAY", modulename );
3224                 if( 0 == search_hash->size ) {
3225                         osrfLogError(
3226                                 OSRF_LOG_MARK,
3227                                 "%s: Invalid predicate structure: empty JSON array",
3228                                 modulename
3229                         );
3230                         buffer_free( sql_buf );
3231                         return NULL;
3232                 }
3233
3234                 unsigned long i = 0;
3235                 while((node = jsonObjectGetIndex( search_hash, i++ ) )) {
3236                         if (first) {
3237                                 first = 0;
3238                         } else {
3239                                 if (opjoin_type == OR_OP_JOIN)
3240                                         buffer_add(sql_buf, " OR ");
3241                                 else
3242                                         buffer_add(sql_buf, " AND ");
3243                         }
3244
3245                         char* subpred = searchWHERE( node, class_info, opjoin_type, ctx );
3246                         if( ! subpred ) {
3247                                 buffer_free( sql_buf );
3248                                 return NULL;
3249                         }
3250
3251                         buffer_fadd(sql_buf, "( %s )", subpred);
3252                         free(subpred);
3253                 }
3254
3255         } else if ( search_hash->type == JSON_HASH ) {
3256                 osrfLogDebug( OSRF_LOG_MARK,
3257                         "%s: In WHERE clause, condition type is JSON_HASH", modulename );
3258                 jsonIterator* search_itr = jsonNewIterator( search_hash );
3259                 if( !jsonIteratorHasNext( search_itr ) ) {
3260                         osrfLogError(
3261                                 OSRF_LOG_MARK,
3262                                 "%s: Invalid predicate structure: empty JSON object",
3263                                 modulename
3264                         );
3265                         jsonIteratorFree( search_itr );
3266                         buffer_free( sql_buf );
3267                         return NULL;
3268                 }
3269
3270                 while ( (node = jsonIteratorNext( search_itr )) ) {
3271
3272                         if (first) {
3273                                 first = 0;
3274                         } else {
3275                                 if (opjoin_type == OR_OP_JOIN)
3276                                         buffer_add(sql_buf, " OR ");
3277                                 else
3278                                         buffer_add(sql_buf, " AND ");
3279                         }
3280
3281                         if ( '+' == search_itr->key[ 0 ] ) {
3282
3283                                 // This plus sign prefixes a class name or other table alias;
3284                                 // make sure the table alias is in scope
3285                                 ClassInfo* alias_info = search_all_alias( search_itr->key + 1 );
3286                                 if( ! alias_info ) {
3287                                         osrfLogError(
3288                                                          OSRF_LOG_MARK,
3289                                                         "%s: Invalid table alias \"%s\" in WHERE clause",
3290                                                         modulename,
3291                                                         search_itr->key + 1
3292                                         );
3293                                         jsonIteratorFree( search_itr );
3294                                         buffer_free( sql_buf );
3295                                         return NULL;
3296                                 }
3297
3298                                 if ( node->type == JSON_STRING ) {
3299                                         // It's the name of a column; make sure it belongs to the class
3300                                         const char* fieldname = jsonObjectGetString( node );
3301                                         if( ! osrfHashGet( alias_info->fields, fieldname ) ) {
3302                                                 osrfLogError(
3303                                                         OSRF_LOG_MARK,
3304                                                         "%s: Invalid column name \"%s\" in WHERE clause "
3305                                                         "for table alias \"%s\"",
3306                                                         modulename,
3307                                                         fieldname,
3308                                                         alias_info->alias
3309                                                 );
3310                                                 jsonIteratorFree( search_itr );
3311                                                 buffer_free( sql_buf );
3312                                                 return NULL;
3313                                         }
3314
3315                                         buffer_fadd(sql_buf, " \"%s\".%s ", alias_info->alias, fieldname );
3316                                 } else {
3317                                         // It's something more complicated
3318                                         char* subpred = searchWHERE( node, alias_info, AND_OP_JOIN, ctx );
3319                                         if( ! subpred ) {
3320                                                 jsonIteratorFree( search_itr );
3321                                                 buffer_free( sql_buf );
3322                                                 return NULL;
3323                                         }
3324
3325                                         buffer_fadd(sql_buf, "( %s )", subpred);
3326                                         free(subpred);
3327                                 }
3328                         } else if ( '-' == search_itr->key[ 0 ] ) {
3329                                 if ( !strcasecmp("-or",search_itr->key) ) {
3330                                         char* subpred = searchWHERE( node, class_info, OR_OP_JOIN, ctx );
3331                                         if( ! subpred ) {
3332                                                 jsonIteratorFree( search_itr );
3333                                                 buffer_free( sql_buf );
3334                                                 return NULL;
3335                                         }
3336
3337                                         buffer_fadd(sql_buf, "( %s )", subpred);
3338                                         free( subpred );
3339                                 } else if ( !strcasecmp("-and",search_itr->key) ) {
3340                                         char* subpred = searchWHERE( node, class_info, AND_OP_JOIN, ctx );
3341                                         if( ! subpred ) {
3342                                                 jsonIteratorFree( search_itr );
3343                                                 buffer_free( sql_buf );
3344                                                 return NULL;
3345                                         }
3346
3347                                         buffer_fadd(sql_buf, "( %s )", subpred);
3348                                         free( subpred );
3349                                 } else if ( !strcasecmp("-not",search_itr->key) ) {
3350                                         char* subpred = searchWHERE( node, class_info, AND_OP_JOIN, ctx );
3351                                         if( ! subpred ) {
3352                                                 jsonIteratorFree( search_itr );
3353                                                 buffer_free( sql_buf );
3354                                                 return NULL;
3355                                         }
3356
3357                                         buffer_fadd(sql_buf, " NOT ( %s )", subpred);
3358                                         free( subpred );
3359                                 } else if ( !strcasecmp("-exists",search_itr->key) ) {
3360                                         char* subpred = buildQuery( ctx, node, SUBSELECT );
3361                                         if( ! subpred ) {
3362                                                 jsonIteratorFree( search_itr );
3363                                                 buffer_free( sql_buf );
3364                                                 return NULL;
3365                                         }
3366
3367                                         buffer_fadd(sql_buf, "EXISTS ( %s )", subpred);
3368                                         free(subpred);
3369                                 } else if ( !strcasecmp("-not-exists",search_itr->key) ) {
3370                                         char* subpred = buildQuery( ctx, node, SUBSELECT );
3371                                         if( ! subpred ) {
3372                                                 jsonIteratorFree( search_itr );
3373                                                 buffer_free( sql_buf );
3374                                                 return NULL;
3375                                         }
3376
3377                                         buffer_fadd(sql_buf, "NOT EXISTS ( %s )", subpred);
3378                                         free(subpred);
3379                                 } else {     // Invalid "minus" operator
3380                                         osrfLogError(
3381                                                          OSRF_LOG_MARK,
3382                                                         "%s: Invalid operator \"%s\" in WHERE clause",
3383                                                         modulename,
3384                                                         search_itr->key
3385                                         );
3386                                         jsonIteratorFree( search_itr );
3387                                         buffer_free( sql_buf );
3388                                         return NULL;
3389                                 }
3390
3391                         } else {
3392
3393                                 const char* class = class_info->class_name;
3394                                 osrfHash* fields = class_info->fields;
3395                                 osrfHash* field = osrfHashGet( fields, search_itr->key );
3396
3397                                 if (!field) {
3398                                         const char* table = class_info->source_def;
3399                                         osrfLogError(
3400                                                 OSRF_LOG_MARK,
3401                                                 "%s: Attempt to reference non-existent column \"%s\" on %s (%s)",
3402                                                 modulename,
3403                                                 search_itr->key,
3404                                                 table ? table : "?",
3405                                                 class ? class : "?"
3406                                         );
3407                                         jsonIteratorFree(search_itr);
3408                                         buffer_free(sql_buf);
3409                                         return NULL;
3410                                 }
3411
3412                                 char* subpred = searchPredicate( class_info, field, node, ctx );
3413                                 if( ! subpred ) {
3414                                         buffer_free(sql_buf);
3415                                         jsonIteratorFree(search_itr);
3416                                         return NULL;
3417                                 }
3418
3419                                 buffer_add( sql_buf, subpred );
3420                                 free(subpred);
3421                         }
3422                 }
3423                 jsonIteratorFree(search_itr);
3424
3425         } else {
3426                 // ERROR ... only hash and array allowed at this level
3427                 char* predicate_string = jsonObjectToJSON( search_hash );
3428                 osrfLogError(
3429                         OSRF_LOG_MARK,
3430                         "%s: Invalid predicate structure: %s",
3431                         modulename,
3432                         predicate_string
3433                 );
3434                 buffer_free(sql_buf);
3435                 free(predicate_string);
3436                 return NULL;
3437         }
3438
3439         return buffer_release(sql_buf);
3440 }
3441
3442 /* Build a JSON_ARRAY of field names for a given table alias
3443 */
3444 static jsonObject* defaultSelectList( const char* table_alias ) {
3445
3446         if( ! table_alias )
3447                 table_alias = "";
3448
3449         ClassInfo* class_info = search_all_alias( table_alias );
3450         if( ! class_info ) {
3451                 osrfLogError(
3452                         OSRF_LOG_MARK,
3453                         "%s: Can't build default SELECT clause for \"%s\"; no such table alias",
3454                         modulename,
3455                         table_alias
3456                 );
3457                 return NULL;
3458         }
3459
3460         jsonObject* array = jsonNewObjectType( JSON_ARRAY );
3461         osrfHash* field_def = NULL;
3462         osrfHashIterator* field_itr = osrfNewHashIterator( class_info->fields );
3463         while( ( field_def = osrfHashIteratorNext( field_itr ) ) ) {
3464                 const char* field_name = osrfHashIteratorKey( field_itr );
3465                 if( ! str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
3466                         jsonObjectPush( array, jsonNewObject( field_name ) );
3467                 }
3468         }
3469         osrfHashIteratorFree( field_itr );
3470
3471         return array;
3472 }
3473
3474 // Translate a jsonObject into a UNION, INTERSECT, or EXCEPT query.
3475 // The jsonObject must be a JSON_HASH with an single entry for "union",
3476 // "intersect", or "except".  The data associated with this key must be an
3477 // array of hashes, each hash being a query.
3478 // Also allowed but currently ignored: entries for "order_by" and "alias".
3479 static char* doCombo( osrfMethodContext* ctx, jsonObject* combo, int flags ) {
3480         // Sanity check
3481         if( ! combo || combo->type != JSON_HASH )
3482                 return NULL;      // should be impossible; validated by caller
3483
3484         const jsonObject* query_array = NULL;   // array of subordinate queries
3485         const char* op = NULL;     // name of operator, e.g. UNION
3486         const char* alias = NULL;  // alias for the query (needed for ORDER BY)
3487         int op_count = 0;          // for detecting conflicting operators
3488         int excepting = 0;         // boolean
3489         int all = 0;               // boolean
3490         jsonObject* order_obj = NULL;
3491
3492         // Identify the elements in the hash
3493         jsonIterator* query_itr = jsonNewIterator( combo );
3494         jsonObject* curr_obj = NULL;
3495         while( (curr_obj = jsonIteratorNext( query_itr ) ) ) {
3496                 if( ! strcmp( "union", query_itr->key ) ) {
3497                         ++op_count;
3498                         op = " UNION ";
3499                         query_array = curr_obj;
3500                 } else if( ! strcmp( "intersect", query_itr->key ) ) {
3501                         ++op_count;
3502                         op = " INTERSECT ";
3503                         query_array = curr_obj;
3504                 } else if( ! strcmp( "except", query_itr->key ) ) {
3505                         ++op_count;
3506                         op = " EXCEPT ";
3507                         excepting = 1;
3508                         query_array = curr_obj;
3509                 } else if( ! strcmp( "order_by", query_itr->key ) ) {
3510                         osrfLogWarning(
3511                                 OSRF_LOG_MARK,
3512                                 "%s: ORDER BY not supported for UNION, INTERSECT, or EXCEPT",
3513                                 modulename
3514                         );
3515                         order_obj = curr_obj;
3516                 } else if( ! strcmp( "alias", query_itr->key ) ) {
3517                         if( curr_obj->type != JSON_STRING ) {
3518                                 jsonIteratorFree( query_itr );
3519                                 return NULL;
3520                         }
3521                         alias = jsonObjectGetString( curr_obj );
3522                 } else if( ! strcmp( "all", query_itr->key ) ) {
3523                         if( obj_is_true( curr_obj ) )
3524                                 all = 1;
3525                 } else {
3526                         if( ctx )
3527                                 osrfAppSessionStatus(
3528                                         ctx->session,
3529                                         OSRF_STATUS_INTERNALSERVERERROR,
3530                                         "osrfMethodException",
3531                                         ctx->request,
3532                                         "Malformed query; unexpected entry in query object"
3533                                 );
3534                         osrfLogError(
3535                                 OSRF_LOG_MARK,
3536                                 "%s: Unexpected entry for \"%s\" in%squery",
3537                                 modulename,
3538                                 query_itr->key,
3539                                 op
3540                         );
3541                         jsonIteratorFree( query_itr );
3542                         return NULL;
3543                 }
3544         }
3545         jsonIteratorFree( query_itr );
3546
3547         // More sanity checks
3548         if( ! query_array ) {
3549                 if( ctx )
3550                         osrfAppSessionStatus(
3551                                 ctx->session,
3552                                 OSRF_STATUS_INTERNALSERVERERROR,
3553                                 "osrfMethodException",
3554                                 ctx->request,
3555                                 "Expected UNION, INTERSECT, or EXCEPT operator not found"
3556                         );
3557                 osrfLogError(
3558                         OSRF_LOG_MARK,
3559                         "%s: Expected UNION, INTERSECT, or EXCEPT operator not found",
3560                         modulename
3561                 );
3562                 return NULL;        // should be impossible...
3563         } else if( op_count > 1 ) {
3564                 if( ctx )
3565                                 osrfAppSessionStatus(
3566                                 ctx->session,
3567                                 OSRF_STATUS_INTERNALSERVERERROR,
3568                                 "osrfMethodException",
3569                                 ctx->request,
3570                                 "Found more than one of UNION, INTERSECT, and EXCEPT in same query"
3571                         );
3572                 osrfLogError(
3573                         OSRF_LOG_MARK,
3574                         "%s: Found more than one of UNION, INTERSECT, and EXCEPT in same query",
3575                         modulename
3576                 );
3577                 return NULL;
3578         } if( query_array->type != JSON_ARRAY ) {
3579                 if( ctx )
3580                                 osrfAppSessionStatus(
3581                                 ctx->session,
3582                                 OSRF_STATUS_INTERNALSERVERERROR,
3583                                 "osrfMethodException",
3584                                 ctx->request,
3585                                 "Malformed query: expected array of queries under UNION, INTERSECT or EXCEPT"
3586                         );
3587                 osrfLogError(
3588                         OSRF_LOG_MARK,
3589                         "%s: Expected JSON_ARRAY of queries for%soperator; found %s",
3590                         modulename,
3591                         op,
3592                         json_type( query_array->type )
3593                 );
3594                 return NULL;
3595         } if( query_array->size < 2 ) {
3596                 if( ctx )
3597                         osrfAppSessionStatus(
3598                                 ctx->session,
3599                                 OSRF_STATUS_INTERNALSERVERERROR,
3600                                 "osrfMethodException",
3601                                 ctx->request,
3602                                 "UNION, INTERSECT or EXCEPT requires multiple queries as operands"
3603                         );
3604                 osrfLogError(
3605                         OSRF_LOG_MARK,
3606                         "%s:%srequires multiple queries as operands",
3607                         modulename,
3608                         op
3609                 );
3610                 return NULL;
3611         } else if( excepting && query_array->size > 2 ) {
3612                 if( ctx )
3613                         osrfAppSessionStatus(
3614                                 ctx->session,
3615                                 OSRF_STATUS_INTERNALSERVERERROR,
3616                                 "osrfMethodException",
3617                                 ctx->request,
3618                                 "EXCEPT operator has too many queries as operands"
3619                         );
3620                 osrfLogError(
3621                         OSRF_LOG_MARK,
3622                         "%s:EXCEPT operator has too many queries as operands",
3623                         modulename
3624                 );
3625                 return NULL;
3626         } else if( order_obj && ! alias ) {
3627                 if( ctx )
3628                         osrfAppSessionStatus(
3629                                 ctx->session,
3630                                 OSRF_STATUS_INTERNALSERVERERROR,
3631                                 "osrfMethodException",
3632                                 ctx->request,
3633                                 "ORDER BY requires an alias for a UNION, INTERSECT, or EXCEPT"
3634                         );
3635                 osrfLogError(
3636                         OSRF_LOG_MARK,
3637                         "%s:ORDER BY requires an alias for a UNION, INTERSECT, or EXCEPT",
3638                         modulename
3639                 );
3640                 return NULL;
3641         }
3642
3643         // So far so good.  Now build the SQL.
3644         growing_buffer* sql = buffer_init( 256 );
3645
3646         // If we nested inside another UNION, INTERSECT, or EXCEPT,
3647         // Add a layer of parentheses
3648         if( flags & SUBCOMBO )
3649                 OSRF_BUFFER_ADD( sql, "( " );
3650
3651         // Traverse the query array.  Each entry should be a hash.
3652         int first = 1;   // boolean
3653         int i = 0;
3654         jsonObject* query = NULL;
3655         while((query = jsonObjectGetIndex( query_array, i++ ) )) {
3656                 if( query->type != JSON_HASH ) {
3657                         if( ctx )
3658                                 osrfAppSessionStatus(
3659                                         ctx->session,
3660                                         OSRF_STATUS_INTERNALSERVERERROR,
3661                                         "osrfMethodException",
3662                                         ctx->request,
3663                                         "Malformed query under UNION, INTERSECT or EXCEPT"
3664                                 );
3665                         osrfLogError(
3666                                 OSRF_LOG_MARK,
3667                                 "%s: Malformed query under%s -- expected JSON_HASH, found %s",
3668                                 modulename,
3669                                 op,
3670                                 json_type( query->type )
3671                         );
3672                         buffer_free( sql );
3673                         return NULL;
3674                 }
3675
3676                 if( first )
3677                         first = 0;
3678                 else {
3679                         OSRF_BUFFER_ADD( sql, op );
3680                         if( all )
3681                                 OSRF_BUFFER_ADD( sql, "ALL " );
3682                 }
3683
3684                 char* query_str = buildQuery( ctx, query, SUBSELECT | SUBCOMBO );
3685                 if( ! query_str ) {
3686                         osrfLogError(
3687                                 OSRF_LOG_MARK,
3688                                 "%s: Error building query under%s",
3689                                 modulename,
3690                                 op
3691                         );
3692                         buffer_free( sql );
3693                         return NULL;
3694                 }
3695
3696                 OSRF_BUFFER_ADD( sql, query_str );
3697         }
3698
3699         if( flags & SUBCOMBO )
3700                 OSRF_BUFFER_ADD_CHAR( sql, ')' );
3701
3702         if ( !(flags & SUBSELECT) )
3703                 OSRF_BUFFER_ADD_CHAR( sql, ';' );
3704
3705         return buffer_release( sql );
3706 }
3707
3708 // Translate a jsonObject into a SELECT, UNION, INTERSECT, or EXCEPT query.
3709 // The jsonObject must be a JSON_HASH with an entry for "from", "union", "intersect",
3710 // or "except" to indicate the type of query.
3711 char* buildQuery( osrfMethodContext* ctx, jsonObject* query, int flags ) {
3712         // Sanity checks
3713         if( ! query ) {
3714                 if( ctx )
3715                         osrfAppSessionStatus(
3716                                 ctx->session,
3717                                 OSRF_STATUS_INTERNALSERVERERROR,
3718                                 "osrfMethodException",
3719                                 ctx->request,
3720                                 "Malformed query; no query object"
3721                         );
3722                 osrfLogError( OSRF_LOG_MARK, "%s: Null pointer to query object", modulename );
3723                 return NULL;
3724         } else if( query->type != JSON_HASH ) {
3725                 if( ctx )
3726                         osrfAppSessionStatus(
3727                                 ctx->session,
3728                                 OSRF_STATUS_INTERNALSERVERERROR,
3729                                 "osrfMethodException",
3730                                 ctx->request,
3731                                 "Malformed query object"
3732                         );
3733                 osrfLogError(
3734                         OSRF_LOG_MARK,
3735                         "%s: Query object is %s instead of JSON_HASH",
3736                         modulename,
3737                         json_type( query->type )
3738                 );
3739                 return NULL;
3740         }
3741
3742         // Determine what kind of query it purports to be, and dispatch accordingly.
3743         if( jsonObjectGetKey( query, "union" ) ||
3744                 jsonObjectGetKey( query, "intersect" ) ||
3745                 jsonObjectGetKey( query, "except" ) ) {
3746                 return doCombo( ctx, query, flags );
3747         } else {
3748                 // It is presumably a SELECT query
3749
3750                 // Push a node onto the stack for the current query.  Every level of
3751                 // subquery gets its own QueryFrame on the Stack.
3752                 push_query_frame();
3753
3754                 // Build an SQL SELECT statement
3755                 char* sql = SELECT(
3756                         ctx,
3757                         jsonObjectGetKey( query, "select" ),
3758                         jsonObjectGetKey( query, "from" ),
3759                         jsonObjectGetKey( query, "where" ),
3760                         jsonObjectGetKey( query, "having" ),
3761                         jsonObjectGetKey( query, "order_by" ),
3762                         jsonObjectGetKey( query, "limit" ),
3763                         jsonObjectGetKey( query, "offset" ),
3764                         flags
3765                 );
3766                 pop_query_frame();
3767                 return sql;
3768         }
3769 }
3770
3771 char* SELECT (
3772                 /* method context */ osrfMethodContext* ctx,
3773
3774                 /* SELECT   */ jsonObject* selhash,
3775                 /* FROM     */ jsonObject* join_hash,
3776                 /* WHERE    */ jsonObject* search_hash,
3777                 /* HAVING   */ jsonObject* having_hash,
3778                 /* ORDER BY */ jsonObject* order_hash,
3779                 /* LIMIT    */ jsonObject* limit,
3780                 /* OFFSET   */ jsonObject* offset,
3781                 /* flags    */ int flags
3782 ) {
3783         const char* locale = osrf_message_get_last_locale();
3784
3785         // general tmp objects
3786         const jsonObject* tmp_const;
3787         jsonObject* selclass = NULL;
3788         jsonObject* snode = NULL;
3789         jsonObject* onode = NULL;
3790
3791         char* string = NULL;
3792         int from_function = 0;
3793         int first = 1;
3794         int gfirst = 1;
3795         //int hfirst = 1;
3796
3797         osrfLogDebug(OSRF_LOG_MARK, "cstore SELECT locale: %s", locale ? locale : "(none)" );
3798
3799         // punt if there's no FROM clause
3800         if (!join_hash || ( join_hash->type == JSON_HASH && !join_hash->size )) {
3801                 osrfLogError(
3802                         OSRF_LOG_MARK,
3803                         "%s: FROM clause is missing or empty",
3804                         modulename
3805                 );
3806                 if( ctx )
3807                         osrfAppSessionStatus(
3808                                 ctx->session,
3809                                 OSRF_STATUS_INTERNALSERVERERROR,
3810                                 "osrfMethodException",
3811                                 ctx->request,
3812                                 "FROM clause is missing or empty in JSON query"
3813                         );
3814                 return NULL;
3815         }
3816
3817         // the core search class
3818         const char* core_class = NULL;
3819
3820         // get the core class -- the only key of the top level FROM clause, or a string
3821         if (join_hash->type == JSON_HASH) {
3822                 jsonIterator* tmp_itr = jsonNewIterator( join_hash );
3823                 snode = jsonIteratorNext( tmp_itr );
3824
3825                 // Populate the current QueryFrame with information
3826                 // about the core class
3827                 if( add_query_core( NULL, tmp_itr->key ) ) {
3828                         if( ctx )
3829                                 osrfAppSessionStatus(
3830                                         ctx->session,
3831                                         OSRF_STATUS_INTERNALSERVERERROR,
3832                                         "osrfMethodException",
3833                                         ctx->request,
3834                                         "Unable to look up core class"
3835                                 );
3836                         return NULL;
3837                 }
3838                 core_class = curr_query->core.class_name;
3839                 join_hash = snode;
3840
3841                 jsonObject* extra = jsonIteratorNext( tmp_itr );
3842
3843                 jsonIteratorFree( tmp_itr );
3844                 snode = NULL;
3845
3846                 // There shouldn't be more than one entry in join_hash
3847                 if( extra ) {
3848                         osrfLogError(
3849                                 OSRF_LOG_MARK,
3850                                 "%s: Malformed FROM clause: extra entry in JSON_HASH",
3851                                 modulename
3852                         );
3853                         if( ctx )
3854                                 osrfAppSessionStatus(
3855                                         ctx->session,
3856                                         OSRF_STATUS_INTERNALSERVERERROR,
3857                                         "osrfMethodException",
3858                                         ctx->request,
3859                                         "Malformed FROM clause in JSON query"
3860                                 );
3861                         return NULL;    // Malformed join_hash; extra entry
3862                 }
3863         } else if (join_hash->type == JSON_ARRAY) {
3864                 // We're selecting from a function, not from a table
3865                 from_function = 1;
3866                 core_class = jsonObjectGetString( jsonObjectGetIndex(join_hash, 0) );
3867                 selhash = NULL;
3868
3869         } else if (join_hash->type == JSON_STRING) {
3870                 // Populate the current QueryFrame with information
3871                 // about the core class
3872                 core_class = jsonObjectGetString( join_hash );
3873                 join_hash = NULL;
3874                 if( add_query_core( NULL, core_class ) ) {
3875                         if( ctx )
3876                                 osrfAppSessionStatus(
3877                                         ctx->session,
3878                                         OSRF_STATUS_INTERNALSERVERERROR,
3879                                         "osrfMethodException",
3880                                         ctx->request,
3881                                         "Unable to look up core class"
3882                                 );
3883                         return NULL;
3884                 }
3885         }
3886         else {
3887                 osrfLogError(
3888                         OSRF_LOG_MARK,
3889                         "%s: FROM clause is unexpected JSON type: %s",
3890                         modulename,
3891                         json_type( join_hash->type )
3892                 );
3893                 if( ctx )
3894                         osrfAppSessionStatus(
3895                                 ctx->session,
3896                                 OSRF_STATUS_INTERNALSERVERERROR,
3897                                 "osrfMethodException",
3898                                 ctx->request,
3899                                 "Ill-formed FROM clause in JSON query"
3900                         );
3901                 return NULL;
3902         }
3903
3904         // Build the join clause, if any, while filling out the list
3905         // of joined classes in the current QueryFrame.
3906         char* join_clause = NULL;
3907         if( join_hash && ! from_function ) {
3908
3909                 join_clause = searchJOIN( join_hash, &curr_query->core );
3910                 if( ! join_clause ) {
3911                         if (ctx)
3912                                 osrfAppSessionStatus(
3913                                         ctx->session,
3914                                         OSRF_STATUS_INTERNALSERVERERROR,
3915                                         "osrfMethodException",
3916                                         ctx->request,
3917                                         "Unable to construct JOIN clause(s)"
3918                                 );
3919                         return NULL;
3920                 }
3921         }
3922
3923         // For in case we don't get a select list
3924         jsonObject* defaultselhash = NULL;
3925
3926         // if there is no select list, build a default select list ...
3927         if (!selhash && !from_function) {
3928                 jsonObject* default_list = defaultSelectList( core_class );
3929                 if( ! default_list ) {
3930                         if (ctx) {
3931                                 osrfAppSessionStatus(
3932                                         ctx->session,
3933                                         OSRF_STATUS_INTERNALSERVERERROR,
3934                                         "osrfMethodException",
3935                                         ctx->request,
3936                                         "Unable to build default SELECT clause in JSON query"
3937                                 );
3938                                 free( join_clause );
3939                                 return NULL;
3940                         }
3941                 }
3942
3943                 selhash = defaultselhash = jsonNewObjectType(JSON_HASH);
3944                 jsonObjectSetKey( selhash, core_class, default_list );
3945         }
3946
3947         // The SELECT clause can be encoded only by a hash
3948         if( !from_function && selhash->type != JSON_HASH ) {
3949                 osrfLogError(
3950                         OSRF_LOG_MARK,
3951                         "%s: Expected JSON_HASH for SELECT clause; found %s",
3952                         modulename,
3953                         json_type( selhash->type )
3954                 );
3955
3956                 if (ctx)
3957                         osrfAppSessionStatus(
3958                                 ctx->session,
3959                                 OSRF_STATUS_INTERNALSERVERERROR,
3960                                 "osrfMethodException",
3961                                 ctx->request,
3962                                 "Malformed SELECT clause in JSON query"
3963                         );
3964                 free( join_clause );
3965                 return NULL;
3966         }
3967
3968         // If you see a null or wild card specifier for the core class, or an
3969         // empty array, replace it with a default SELECT list
3970         tmp_const = jsonObjectGetKeyConst( selhash, core_class );
3971         if ( tmp_const ) {
3972                 int default_needed = 0;   // boolean
3973                 if( JSON_STRING == tmp_const->type
3974                         && !strcmp( "*", jsonObjectGetString( tmp_const ) ))
3975                                 default_needed = 1;
3976                 else if( JSON_NULL == tmp_const->type )
3977                         default_needed = 1;
3978
3979                 if( default_needed ) {
3980                         // Build a default SELECT list
3981                         jsonObject* default_list = defaultSelectList( core_class );
3982                         if( ! default_list ) {
3983                                 if (ctx) {
3984                                         osrfAppSessionStatus(
3985                                                 ctx->session,
3986                                                 OSRF_STATUS_INTERNALSERVERERROR,
3987                                                 "osrfMethodException",
3988                                                 ctx->request,
3989                                                 "Can't build default SELECT clause in JSON query"
3990                                         );
3991                                         free( join_clause );
3992                                         return NULL;
3993                                 }
3994                         }
3995
3996                         jsonObjectSetKey( selhash, core_class, default_list );
3997                 }
3998         }
3999
4000         // temp buffers for the SELECT list and GROUP BY clause
4001         growing_buffer* select_buf = buffer_init(128);
4002         growing_buffer* group_buf = buffer_init(128);
4003
4004         int aggregate_found = 0;     // boolean
4005
4006         // Build a select list
4007         if(from_function)   // From a function we select everything
4008                 OSRF_BUFFER_ADD_CHAR( select_buf, '*' );
4009         else {
4010
4011                 // Build the SELECT list as SQL
4012             int sel_pos = 1;
4013             first = 1;
4014             gfirst = 1;
4015             jsonIterator* selclass_itr = jsonNewIterator( selhash );
4016             while ( (selclass = jsonIteratorNext( selclass_itr )) ) {    // For each class
4017
4018                         const char* cname = selclass_itr->key;
4019
4020                         // Make sure the target relation is in the FROM clause.
4021
4022                         // At this point join_hash is a step down from the join_hash we
4023                         // received as a parameter.  If the original was a JSON_STRING,
4024                         // then json_hash is now NULL.  If the original was a JSON_HASH,
4025                         // then json_hash is now the first (and only) entry in it,
4026                         // denoting the core class.  We've already excluded the
4027                         // possibility that the original was a JSON_ARRAY, because in
4028                         // that case from_function would be non-NULL, and we wouldn't
4029                         // be here.
4030
4031                         // If the current table alias isn't in scope, bail out
4032                         ClassInfo* class_info = search_alias( cname );
4033                         if( ! class_info ) {
4034                                 osrfLogError(
4035                                         OSRF_LOG_MARK,
4036                                         "%s: SELECT clause references class not in FROM clause: \"%s\"",
4037                                         modulename,
4038                                         cname
4039                                 );
4040                                 if( ctx )
4041                                         osrfAppSessionStatus(
4042                                                 ctx->session,
4043                                                 OSRF_STATUS_INTERNALSERVERERROR,
4044                                                 "osrfMethodException",
4045                                                 ctx->request,
4046                                                 "Selected class not in FROM clause in JSON query"
4047                                         );
4048                                 jsonIteratorFree( selclass_itr );
4049                                 buffer_free( select_buf );
4050                                 buffer_free( group_buf );
4051                                 if( defaultselhash )
4052                                         jsonObjectFree( defaultselhash );
4053                                 free( join_clause );
4054                                 return NULL;
4055                         }
4056
4057                         if( selclass->type != JSON_ARRAY ) {
4058                                 osrfLogError(
4059                                         OSRF_LOG_MARK,
4060                                         "%s: Malformed SELECT list for class \"%s\"; not an array",
4061                                         modulename,
4062                                         cname
4063                                 );
4064                                 if( ctx )
4065                                         osrfAppSessionStatus(
4066                                                 ctx->session,
4067                                                 OSRF_STATUS_INTERNALSERVERERROR,
4068                                                 "osrfMethodException",
4069                                                 ctx->request,
4070                                                 "Selected class not in FROM clause in JSON query"
4071                                         );
4072
4073                                 jsonIteratorFree( selclass_itr );
4074                                 buffer_free( select_buf );
4075                                 buffer_free( group_buf );
4076                                 if( defaultselhash )
4077                                         jsonObjectFree( defaultselhash );
4078                                 free( join_clause );
4079                                 return NULL;
4080                         }
4081
4082                         // Look up some attributes of the current class
4083                         osrfHash* idlClass = class_info->class_def;
4084                         osrfHash* class_field_set = class_info->fields;
4085                         const char* class_pkey = osrfHashGet( idlClass, "primarykey" );
4086                         const char* class_tname = osrfHashGet( idlClass, "tablename" );
4087
4088                         if( 0 == selclass->size ) {
4089                                 osrfLogWarning(
4090                                         OSRF_LOG_MARK,
4091                                         "%s: No columns selected from \"%s\"",
4092                                         modulename,
4093                                         cname
4094                                 );
4095                         }
4096
4097                         // stitch together the column list for the current table alias...
4098                         unsigned long field_idx = 0;
4099                         jsonObject* selfield = NULL;
4100                         while((selfield = jsonObjectGetIndex( selclass, field_idx++ ) )) {
4101
4102                                 // If we need a separator comma, add one
4103                                 if (first) {
4104                                         first = 0;
4105                                 } else {
4106                                         OSRF_BUFFER_ADD_CHAR( select_buf, ',' );
4107                                 }
4108
4109                                 // if the field specification is a string, add it to the list
4110                                 if (selfield->type == JSON_STRING) {
4111
4112                                         // Look up the field in the IDL
4113                                         const char* col_name = jsonObjectGetString( selfield );
4114                                         osrfHash* field_def = osrfHashGet( class_field_set, col_name );
4115                                         if ( !field_def ) {
4116                                                 // No such field in current class
4117                                                 osrfLogError(
4118                                                         OSRF_LOG_MARK,
4119                                                         "%s: Selected column \"%s\" not defined in IDL for class \"%s\"",
4120                                                         modulename,
4121                                                         col_name,
4122                                                         cname
4123                                                 );
4124                                                 if( ctx )
4125                                                         osrfAppSessionStatus(
4126                                                                 ctx->session,
4127                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4128                                                                 "osrfMethodException",
4129                                                                 ctx->request,
4130                                                                 "Selected column not defined in JSON query"
4131                                                         );
4132                                                 jsonIteratorFree( selclass_itr );
4133                                                 buffer_free( select_buf );
4134                                                 buffer_free( group_buf );
4135                                                 if( defaultselhash )
4136                                                         jsonObjectFree( defaultselhash );
4137                                                 free( join_clause );
4138                                                 return NULL;
4139                                         } else if ( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4140                                                 // Virtual field not allowed
4141                                                 osrfLogError(
4142                                                         OSRF_LOG_MARK,
4143                                                         "%s: Selected column \"%s\" for class \"%s\" is virtual",
4144                                                         modulename,
4145                                                         col_name,
4146                                                         cname
4147                                                 );
4148                                                 if( ctx )
4149                                                         osrfAppSessionStatus(
4150                                                                 ctx->session,
4151                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4152                                                                 "osrfMethodException",
4153                                                                 ctx->request,
4154                                                                 "Selected column may not be virtual in JSON query"
4155                                                         );
4156                                                 jsonIteratorFree( selclass_itr );
4157                                                 buffer_free( select_buf );
4158                                                 buffer_free( group_buf );
4159                                                 if( defaultselhash )
4160                                                         jsonObjectFree( defaultselhash );
4161                                                 free( join_clause );
4162                                                 return NULL;
4163                                         }
4164
4165                                         if (locale) {
4166                                                 const char* i18n;
4167                                                 if (flags & DISABLE_I18N)
4168                                                         i18n = NULL;
4169                                                 else
4170                                                         i18n = osrfHashGet(field_def, "i18n");
4171
4172                                                 if( str_is_true( i18n ) ) {
4173                                                         buffer_fadd( select_buf, " oils_i18n_xlate('%s', '%s', '%s', "
4174                                                                 "'%s', \"%s\".%s::TEXT, '%s') AS \"%s\"",
4175                                                                 class_tname, cname, col_name, class_pkey,
4176                                                                 cname, class_pkey, locale, col_name );
4177                                                 } else {
4178                                                         buffer_fadd(select_buf, " \"%s\".%s AS \"%s\"",
4179                                                                 cname, col_name, col_name );
4180                                                 }
4181                                         } else {
4182                                                 buffer_fadd(select_buf, " \"%s\".%s AS \"%s\"",
4183                                                                 cname, col_name, col_name );
4184                                         }
4185
4186                                 // ... but it could be an object, in which case we check for a Field Transform
4187                                 } else if (selfield->type == JSON_HASH) {
4188
4189                                         const char* col_name = jsonObjectGetString(
4190                                                         jsonObjectGetKeyConst( selfield, "column" ) );
4191
4192                                         // Get the field definition from the IDL
4193                                         osrfHash* field_def = osrfHashGet( class_field_set, col_name );
4194                                         if ( !field_def ) {
4195                                                 // No such field in current class
4196                                                 osrfLogError(
4197                                                         OSRF_LOG_MARK,
4198                                                         "%s: Selected column \"%s\" is not defined in IDL for class \"%s\"",
4199                                                         modulename,
4200                                                         col_name,
4201                                                         cname
4202                                                 );
4203                                                 if( ctx )
4204                                                         osrfAppSessionStatus(
4205                                                                 ctx->session,
4206                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4207                                                                 "osrfMethodException",
4208                                                                 ctx->request,
4209                                                                 "Selected column is not defined in JSON query"
4210                                                         );
4211                                                 jsonIteratorFree( selclass_itr );
4212                                                 buffer_free( select_buf );
4213                                                 buffer_free( group_buf );
4214                                                 if( defaultselhash )
4215                                                         jsonObjectFree( defaultselhash );
4216                                                 free( join_clause );
4217                                                 return NULL;
4218                                         } else if ( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4219                                                 // No such field in current class
4220                                                 osrfLogError(
4221                                                         OSRF_LOG_MARK,
4222                                                         "%s: Selected column \"%s\" is virtual for class \"%s\"",
4223                                                         modulename,
4224                                                         col_name,
4225                                                         cname
4226                                                 );
4227                                                 if( ctx )
4228                                                         osrfAppSessionStatus(
4229                                                                 ctx->session,
4230                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4231                                                                 "osrfMethodException",
4232                                                                 ctx->request,
4233                                                                 "Selected column is virtual in JSON query"
4234                                                         );
4235                                                 jsonIteratorFree( selclass_itr );
4236                                                 buffer_free( select_buf );
4237                                                 buffer_free( group_buf );
4238                                                 if( defaultselhash )
4239                                                         jsonObjectFree( defaultselhash );
4240                                                 free( join_clause );
4241                                                 return NULL;
4242                                         }
4243
4244                                         // Decide what to use as a column alias
4245                                         const char* _alias;
4246                                         if ((tmp_const = jsonObjectGetKeyConst( selfield, "alias" ))) {
4247                                                 _alias = jsonObjectGetString( tmp_const );
4248                                         } else {         // Use field name as the alias
4249                                                 _alias = col_name;
4250                                         }
4251
4252                                         if (jsonObjectGetKeyConst( selfield, "transform" )) {
4253                                                 char* transform_str = searchFieldTransform(
4254                                                         class_info->alias, field_def, selfield );
4255                                                 if( transform_str ) {
4256                                                         buffer_fadd(select_buf, " %s AS \"%s\"", transform_str, _alias);
4257                                                         free(transform_str);
4258                                                 } else {
4259                                                         if( ctx )
4260                                                                 osrfAppSessionStatus(
4261                                                                         ctx->session,
4262                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4263                                                                         "osrfMethodException",
4264                                                                         ctx->request,
4265                                                                         "Unable to generate transform function in JSON query"
4266                                                                 );
4267                                                         jsonIteratorFree( selclass_itr );
4268                                                         buffer_free( select_buf );
4269                                                         buffer_free( group_buf );
4270                                                         if( defaultselhash )
4271                                                                 jsonObjectFree( defaultselhash );
4272                                                         free( join_clause );
4273                                                         return NULL;
4274                                                 }
4275                                         } else {
4276
4277                                                 if (locale) {
4278                                                         const char* i18n;
4279                                                         if (flags & DISABLE_I18N)
4280                                                                 i18n = NULL;
4281                                                         else
4282                                                                 i18n = osrfHashGet(field_def, "i18n");
4283
4284                                                         if( str_is_true( i18n ) ) {
4285                                                                 buffer_fadd( select_buf,
4286                                                                         " oils_i18n_xlate('%s', '%s', '%s', '%s', "
4287                                                                         "\"%s\".%s::TEXT, '%s') AS \"%s\"",
4288                                                                         class_tname, cname, col_name, class_pkey, cname,
4289                                                                         class_pkey, locale, _alias);
4290                                                         } else {
4291                                                                 buffer_fadd( select_buf, " \"%s\".%s AS \"%s\"",
4292                                                                         cname, col_name, _alias );
4293                                                         }
4294                                                 } else {
4295                                                         buffer_fadd( select_buf, " \"%s\".%s AS \"%s\"",
4296                                                                 cname, col_name, _alias);
4297                                                 }
4298                                         }
4299                                 }
4300                                 else {
4301                                         osrfLogError(
4302                                                 OSRF_LOG_MARK,
4303                                                 "%s: Selected item is unexpected JSON type: %s",
4304                                                 modulename,
4305                                                 json_type( selfield->type )
4306                                         );
4307                                         if( ctx )
4308                                                 osrfAppSessionStatus(
4309                                                         ctx->session,
4310                                                         OSRF_STATUS_INTERNALSERVERERROR,
4311                                                         "osrfMethodException",
4312                                                         ctx->request,
4313                                                         "Ill-formed SELECT item in JSON query"
4314                                                 );
4315                                         jsonIteratorFree( selclass_itr );
4316                                         buffer_free( select_buf );
4317                                         buffer_free( group_buf );
4318                                         if( defaultselhash )
4319                                                 jsonObjectFree( defaultselhash );
4320                                         free( join_clause );
4321                                         return NULL;
4322                                 }
4323
4324                                 const jsonObject* agg_obj = jsonObjectGetKey( selfield, "aggregate" );
4325                                 if( obj_is_true( agg_obj ) )
4326                                         aggregate_found = 1;
4327                                 else {
4328                                         // Append a comma (except for the first one)
4329                                         // and add the column to a GROUP BY clause
4330                                         if (gfirst)
4331                                                 gfirst = 0;
4332                                         else
4333                                                 OSRF_BUFFER_ADD_CHAR( group_buf, ',' );
4334
4335                                         buffer_fadd(group_buf, " %d", sel_pos);
4336                                 }
4337
4338 #if 0
4339                             if (is_agg->size || (flags & SELECT_DISTINCT)) {
4340
4341                                         const jsonObject* aggregate_obj = jsonObjectGetKey( selfield, "aggregate" );
4342                                     if ( ! obj_is_true( aggregate_obj ) ) {
4343                                             if (gfirst) {
4344                                                     gfirst = 0;
4345                                             } else {
4346                                                         OSRF_BUFFER_ADD_CHAR( group_buf, ',' );
4347                                             }
4348
4349                                             buffer_fadd(group_buf, " %d", sel_pos);
4350
4351                                         /*
4352                                     } else if (is_agg = jsonObjectGetKey( selfield, "having" )) {
4353                                             if (gfirst) {
4354                                                     gfirst = 0;
4355                                             } else {
4356                                                         OSRF_BUFFER_ADD_CHAR( group_buf, ',' );
4357                                             }
4358
4359                                             _column = searchFieldTransform(class_info->alias, field, selfield);
4360                                                 OSRF_BUFFER_ADD_CHAR(group_buf, ' ');
4361                                                 OSRF_BUFFER_ADD(group_buf, _column);
4362                                             _column = searchFieldTransform(class_info->alias, field, selfield);
4363                                         */
4364                                     }
4365                             }
4366 #endif
4367
4368                                 sel_pos++;
4369                         } // end while -- iterating across SELECT columns
4370
4371                 } // end while -- iterating across classes
4372
4373                 jsonIteratorFree(selclass_itr);
4374         }
4375
4376
4377         char* col_list = buffer_release(select_buf);
4378
4379         // Make sure the SELECT list isn't empty.  This can happen, for example,
4380         // if we try to build a default SELECT clause from a non-core table.
4381
4382         if( ! *col_list ) {
4383                 osrfLogError( OSRF_LOG_MARK, "%s: SELECT clause is empty", modulename );
4384                 if (ctx)
4385                         osrfAppSessionStatus(
4386                                 ctx->session,
4387                                 OSRF_STATUS_INTERNALSERVERERROR,
4388                                 "osrfMethodException",
4389                                 ctx->request,
4390                                 "SELECT list is empty"
4391                 );
4392                 free( col_list );
4393                 buffer_free( group_buf );
4394                 if( defaultselhash )
4395                         jsonObjectFree( defaultselhash );
4396                 free( join_clause );
4397                 return NULL;
4398         }
4399
4400         char* table = NULL;
4401         if (from_function) table = searchValueTransform(join_hash);
4402         else table = strdup( curr_query->core.source_def );
4403
4404         if( !table ) {
4405                 if (ctx)
4406                         osrfAppSessionStatus(
4407                                 ctx->session,
4408                                 OSRF_STATUS_INTERNALSERVERERROR,
4409                                 "osrfMethodException",
4410                                 ctx->request,
4411                                 "Unable to identify table for core class"
4412                         );
4413                 free( col_list );
4414                 buffer_free( group_buf );
4415                 if( defaultselhash )
4416                         jsonObjectFree( defaultselhash );
4417                 free( join_clause );
4418                 return NULL;
4419         }
4420
4421         // Put it all together
4422         growing_buffer* sql_buf = buffer_init(128);
4423         buffer_fadd(sql_buf, "SELECT %s FROM %s AS \"%s\" ", col_list, table, core_class );
4424         free(col_list);
4425         free(table);
4426
4427         // Append the join clause, if any
4428         if( join_clause ) {
4429                 buffer_add(sql_buf, join_clause);
4430                 free(join_clause);
4431         }
4432
4433         char* order_by_list = NULL;
4434         char* having_buf = NULL;
4435
4436         if (!from_function) {
4437
4438                 // Build a WHERE clause, if there is one
4439                 if ( search_hash ) {
4440                         buffer_add(sql_buf, " WHERE ");
4441
4442                         // and it's on the WHERE clause
4443                         char* pred = searchWHERE( search_hash, &curr_query->core, AND_OP_JOIN, ctx );
4444                         if ( ! pred ) {
4445                                 if (ctx) {
4446                                         osrfAppSessionStatus(
4447                                                 ctx->session,
4448                                                 OSRF_STATUS_INTERNALSERVERERROR,
4449                                                 "osrfMethodException",
4450                                                 ctx->request,
4451                                                 "Severe query error in WHERE predicate -- see error log for more details"
4452                                         );
4453                                 }
4454                                 buffer_free(group_buf);
4455                                 buffer_free(sql_buf);
4456                                 if (defaultselhash)
4457                                         jsonObjectFree(defaultselhash);
4458                                 return NULL;
4459                         }
4460
4461                         buffer_add(sql_buf, pred);
4462                         free(pred);
4463                 }
4464
4465                 // Build a HAVING clause, if there is one
4466                 if ( having_hash ) {
4467
4468                         // and it's on the the WHERE clause
4469                         having_buf = searchWHERE( having_hash, &curr_query->core, AND_OP_JOIN, ctx );
4470
4471                         if( ! having_buf ) {
4472                                 if (ctx) {
4473                                                 osrfAppSessionStatus(
4474                                                 ctx->session,
4475                                                 OSRF_STATUS_INTERNALSERVERERROR,
4476                                                 "osrfMethodException",
4477                                                 ctx->request,
4478                                                 "Severe query error in HAVING predicate -- see error log for more details"
4479                                         );
4480                                 }
4481                                 buffer_free(group_buf);
4482                                 buffer_free(sql_buf);
4483                                 if (defaultselhash)
4484                                         jsonObjectFree(defaultselhash);
4485                                 return NULL;
4486                         }
4487                 }
4488
4489                 growing_buffer* order_buf = NULL;  // to collect ORDER BY list
4490
4491                 // Build an ORDER BY clause, if there is one
4492                 if( NULL == order_hash )
4493                         ;  // No ORDER BY? do nothing
4494                 else if( JSON_ARRAY == order_hash->type ) {
4495                         // Array of field specifications, each specification being a
4496                         // hash to define the class, field, and other details
4497                         int order_idx = 0;
4498                         jsonObject* order_spec;
4499                         while( (order_spec = jsonObjectGetIndex( order_hash, order_idx++ ) ) ) {
4500
4501                                 if( JSON_HASH != order_spec->type ) {
4502                                         osrfLogError(OSRF_LOG_MARK,
4503                                                  "%s: Malformed field specification in ORDER BY clause; expected JSON_HASH, found %s",
4504                                                 modulename, json_type( order_spec->type ) );
4505                                         if( ctx )
4506                                                 osrfAppSessionStatus(
4507                                                          ctx->session,
4508                                                         OSRF_STATUS_INTERNALSERVERERROR,
4509                                                         "osrfMethodException",
4510                                                         ctx->request,
4511                                                         "Malformed ORDER BY clause -- see error log for more details"
4512                                                 );
4513                                         buffer_free( order_buf );
4514                                         free(having_buf);
4515                                         buffer_free(group_buf);
4516                                         buffer_free(sql_buf);
4517                                         if (defaultselhash)
4518                                                 jsonObjectFree(defaultselhash);
4519                                         return NULL;
4520                                 }
4521
4522                                 const char* class_alias =
4523                                                 jsonObjectGetString( jsonObjectGetKeyConst( order_spec, "class" ) );
4524                                 const char* field =
4525                                                 jsonObjectGetString( jsonObjectGetKeyConst( order_spec, "field" ) );
4526
4527                                 if ( order_buf )
4528                                         OSRF_BUFFER_ADD(order_buf, ", ");
4529                                 else
4530                                         order_buf = buffer_init(128);
4531
4532                                 if( !field || !class_alias ) {
4533                                         osrfLogError( OSRF_LOG_MARK,
4534                                                 "%s: Missing class or field name in field specification "
4535                                                 "of ORDER BY clause",
4536                                                 modulename );
4537                                         if( ctx )
4538                                                 osrfAppSessionStatus(
4539                                                         ctx->session,
4540                                                         OSRF_STATUS_INTERNALSERVERERROR,
4541                                                         "osrfMethodException",
4542                                                         ctx->request,
4543                                                         "Malformed ORDER BY clause -- see error log for more details"
4544                                                 );
4545                                         buffer_free( order_buf );
4546                                         free(having_buf);
4547                                         buffer_free(group_buf);
4548                                         buffer_free(sql_buf);
4549                                         if (defaultselhash)
4550                                                 jsonObjectFree(defaultselhash);
4551                                         return NULL;
4552                                 }
4553
4554                                 ClassInfo* order_class_info = search_alias( class_alias );
4555                                 if( ! order_class_info ) {
4556                                         osrfLogError(OSRF_LOG_MARK, "%s: ORDER BY clause references class \"%s\" "
4557                                                         "not in FROM clause", modulename, class_alias );
4558                                         if( ctx )
4559                                                 osrfAppSessionStatus(
4560                                                         ctx->session,
4561                                                         OSRF_STATUS_INTERNALSERVERERROR,
4562                                                         "osrfMethodException",
4563                                                         ctx->request,
4564                                                         "Invalid class referenced in ORDER BY clause -- "
4565                                                         "see error log for more details"
4566                                                 );
4567                                         free(having_buf);
4568                                         buffer_free(group_buf);
4569                                         buffer_free(sql_buf);
4570                                         if (defaultselhash)
4571                                                 jsonObjectFree(defaultselhash);
4572                                         return NULL;
4573                                 }
4574
4575                                 osrfHash* field_def = osrfHashGet( order_class_info->fields, field );
4576                                 if( !field_def ) {
4577                                         osrfLogError( OSRF_LOG_MARK,
4578                                                 "%s: Invalid field \"%s\".%s referenced in ORDER BY clause",
4579                                                 modulename, class_alias, field );
4580                                         if( ctx )
4581                                                 osrfAppSessionStatus(
4582                                                         ctx->session,
4583                                                         OSRF_STATUS_INTERNALSERVERERROR,
4584                                                         "osrfMethodException",
4585                                                         ctx->request,
4586                                                         "Invalid field referenced in ORDER BY clause -- "
4587                                                         "see error log for more details"
4588                                                 );
4589                                         free(having_buf);
4590                                         buffer_free(group_buf);
4591                                         buffer_free(sql_buf);
4592                                         if (defaultselhash)
4593                                                 jsonObjectFree(defaultselhash);
4594                                         return NULL;
4595                                 } else if( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4596                                         osrfLogError(OSRF_LOG_MARK, "%s: Virtual field \"%s\" in ORDER BY clause",
4597                                                                  modulename, field );
4598                                         if( ctx )
4599                                                 osrfAppSessionStatus(
4600                                                         ctx->session,
4601                                                         OSRF_STATUS_INTERNALSERVERERROR,
4602                                                         "osrfMethodException",
4603                                                         ctx->request,
4604                                                         "Virtual field in ORDER BY clause -- see error log for more details"
4605                                                 );
4606                                         buffer_free( order_buf );
4607                                         free(having_buf);
4608                                         buffer_free(group_buf);
4609                                         buffer_free(sql_buf);
4610                                         if (defaultselhash)
4611                                                 jsonObjectFree(defaultselhash);
4612                                         return NULL;
4613                                 }
4614
4615                                 if( jsonObjectGetKeyConst( order_spec, "transform" ) ) {
4616                                         char* transform_str = searchFieldTransform(
4617                                                         class_alias, field_def, order_spec );
4618                                         if( ! transform_str ) {
4619                                                 if( ctx )
4620                                                         osrfAppSessionStatus(
4621                                                                 ctx->session,
4622                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4623                                                                 "osrfMethodException",
4624                                                                 ctx->request,
4625                                                                 "Severe query error in ORDER BY clause -- "
4626                                                                 "see error log for more details"
4627                                                         );
4628                                                 buffer_free( order_buf );
4629                                                 free(having_buf);
4630                                                 buffer_free(group_buf);
4631                                                 buffer_free(sql_buf);
4632                                                 if (defaultselhash)
4633                                                         jsonObjectFree(defaultselhash);
4634                                                 return NULL;
4635                                         }
4636
4637                                         OSRF_BUFFER_ADD( order_buf, transform_str );
4638                                         free( transform_str );
4639                                 }
4640                                 else
4641                                         buffer_fadd( order_buf, "\"%s\".%s", class_alias, field );
4642
4643                                 const char* direction =
4644                                                 jsonObjectGetString( jsonObjectGetKeyConst( order_spec, "direction" ) );
4645                                 if( direction ) {
4646                                         if( direction[ 0 ] || 'D' == direction[ 0 ] )
4647                                                 OSRF_BUFFER_ADD( order_buf, " DESC" );
4648                                         else
4649                                                 OSRF_BUFFER_ADD( order_buf, " ASC" );
4650                                 }
4651                         }
4652                 } else if( JSON_HASH == order_hash->type ) {
4653                         // This hash is keyed on class alias.  Each class has either
4654                         // an array of field names or a hash keyed on field name.
4655                         jsonIterator* class_itr = jsonNewIterator( order_hash );
4656                         while ( (snode = jsonIteratorNext( class_itr )) ) {
4657
4658                                 ClassInfo* order_class_info = search_alias( class_itr->key );
4659                                 if( ! order_class_info ) {
4660                                         osrfLogError(OSRF_LOG_MARK,
4661                                                 "%s: Invalid class \"%s\" referenced in ORDER BY clause",
4662                                                 modulename, class_itr->key );
4663                                         if( ctx )
4664                                                 osrfAppSessionStatus(
4665                                                         ctx->session,
4666                                                         OSRF_STATUS_INTERNALSERVERERROR,
4667                                                         "osrfMethodException",
4668                                                         ctx->request,
4669                                                         "Invalid class referenced in ORDER BY clause -- "
4670                                                         "see error log for more details"
4671                                                 );
4672                                         jsonIteratorFree( class_itr );
4673                                         buffer_free( order_buf );
4674                                         free(having_buf);
4675                                         buffer_free(group_buf);
4676                                         buffer_free(sql_buf);
4677                                         if (defaultselhash)
4678                                                 jsonObjectFree(defaultselhash);
4679                                         return NULL;
4680                                 }
4681
4682                                 osrfHash* field_list_def = order_class_info->fields;
4683
4684                                 if ( snode->type == JSON_HASH ) {
4685
4686                                         // Hash is keyed on field names from the current class.  For each field
4687                                         // there is another layer of hash to define the sorting details, if any,
4688                                         // or a string to indicate direction of sorting.
4689                                         jsonIterator* order_itr = jsonNewIterator( snode );
4690                                         while ( (onode = jsonIteratorNext( order_itr )) ) {
4691
4692                                                 osrfHash* field_def = osrfHashGet( field_list_def, order_itr->key );
4693                                                 if( !field_def ) {
4694                                                         osrfLogError( OSRF_LOG_MARK,
4695                                                                 "%s: Invalid field \"%s\" in ORDER BY clause",
4696                                                                 modulename, order_itr->key );
4697                                                         if( ctx )
4698                                                                 osrfAppSessionStatus(
4699                                                                         ctx->session,
4700                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4701                                                                         "osrfMethodException",
4702                                                                         ctx->request,
4703                                                                         "Invalid field in ORDER BY clause -- "
4704                                                                         "see error log for more details"
4705                                                                 );
4706                                                         jsonIteratorFree( order_itr );
4707                                                         jsonIteratorFree( class_itr );
4708                                                         buffer_free( order_buf );
4709                                                         free(having_buf);
4710                                                         buffer_free(group_buf);
4711                                                         buffer_free(sql_buf);
4712                                                         if (defaultselhash)
4713                                                                 jsonObjectFree(defaultselhash);
4714                                                         return NULL;
4715                                                 } else if( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4716                                                         osrfLogError( OSRF_LOG_MARK,
4717                                                                 "%s: Virtual field \"%s\" in ORDER BY clause",
4718                                                                 modulename, order_itr->key );
4719                                                         if( ctx )
4720                                                                 osrfAppSessionStatus(
4721                                                                         ctx->session,
4722                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4723                                                                         "osrfMethodException",
4724                                                                         ctx->request,
4725                                                                         "Virtual field in ORDER BY clause -- "
4726                                                                         "see error log for more details"
4727                                                         );
4728                                                         jsonIteratorFree( order_itr );
4729                                                         jsonIteratorFree( class_itr );
4730                                                         buffer_free( order_buf );
4731                                                         free(having_buf);
4732                                                         buffer_free(group_buf);
4733                                                         buffer_free(sql_buf);
4734                                                         if (defaultselhash)
4735                                                                 jsonObjectFree(defaultselhash);
4736                                                         return NULL;
4737                                                 }
4738
4739                                                 const char* direction = NULL;
4740                                                 if ( onode->type == JSON_HASH ) {
4741                                                         if ( jsonObjectGetKeyConst( onode, "transform" ) ) {
4742                                                                 string = searchFieldTransform(
4743                                                                         class_itr->key,
4744                                                                         osrfHashGet( field_list_def, order_itr->key ),
4745                                                                         onode
4746                                                                 );
4747                                                                 if( ! string ) {
4748                                                                         if( ctx ) osrfAppSessionStatus(
4749                                                                                 ctx->session,
4750                                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4751                                                                                 "osrfMethodException",
4752                                                                                 ctx->request,
4753                                                                                 "Severe query error in ORDER BY clause -- "
4754                                                                                 "see error log for more details"
4755                                                                         );
4756                                                                         jsonIteratorFree( order_itr );
4757                                                                         jsonIteratorFree( class_itr );
4758                                                                         free(having_buf);
4759                                                                         buffer_free(group_buf);
4760                                                                         buffer_free(order_buf);
4761                                                                         buffer_free(sql_buf);
4762                                                                         if (defaultselhash)
4763                                                                                 jsonObjectFree(defaultselhash);
4764                                                                         return NULL;
4765                                                                 }
4766                                                         } else {
4767                                                                 growing_buffer* field_buf = buffer_init(16);
4768                                                                 buffer_fadd( field_buf, "\"%s\".%s",
4769                                                                         class_itr->key, order_itr->key );
4770                                                                 string = buffer_release(field_buf);
4771                                                         }
4772
4773                                                         if ( (tmp_const = jsonObjectGetKeyConst( onode, "direction" )) ) {
4774                                                                 const char* dir = jsonObjectGetString(tmp_const);
4775                                                                 if (!strncasecmp(dir, "d", 1)) {
4776                                                                         direction = " DESC";
4777                                                                 } else {
4778                                                                         direction = " ASC";
4779                                                                 }
4780                                                         }
4781
4782                                                 } else if ( JSON_NULL == onode->type || JSON_ARRAY == onode->type ) {
4783                                                         osrfLogError( OSRF_LOG_MARK,
4784                                                                 "%s: Expected JSON_STRING in ORDER BY clause; found %s",
4785                                                                 modulename, json_type( onode->type ) );
4786                                                         if( ctx )
4787                                                                 osrfAppSessionStatus(
4788                                                                         ctx->session,
4789                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4790                                                                         "osrfMethodException",
4791                                                                         ctx->request,
4792                                                                         "Malformed ORDER BY clause -- see error log for more details"
4793                                                                 );
4794                                                         jsonIteratorFree( order_itr );
4795                                                         jsonIteratorFree( class_itr );
4796                                                         free(having_buf);
4797                                                         buffer_free(group_buf);
4798                                                         buffer_free(order_buf);
4799                                                         buffer_free(sql_buf);
4800                                                         if (defaultselhash)
4801                                                                 jsonObjectFree(defaultselhash);
4802                                                         return NULL;
4803
4804                                                 } else {
4805                                                         string = strdup(order_itr->key);
4806                                                         const char* dir = jsonObjectGetString(onode);
4807                                                         if (!strncasecmp(dir, "d", 1)) {
4808                                                                 direction = " DESC";
4809                                                         } else {
4810                                                                 direction = " ASC";
4811                                                         }
4812                                                 }
4813
4814                                                 if ( order_buf )
4815                                                         OSRF_BUFFER_ADD(order_buf, ", ");
4816                                                 else
4817                                                         order_buf = buffer_init(128);
4818
4819                                                 OSRF_BUFFER_ADD(order_buf, string);
4820                                                 free(string);
4821
4822                                                 if (direction) {
4823                                                          OSRF_BUFFER_ADD(order_buf, direction);
4824                                                 }
4825
4826                                         } // end while
4827                                         jsonIteratorFree(order_itr);
4828
4829                                 } else if ( snode->type == JSON_ARRAY ) {
4830
4831                                         // Array is a list of fields from the current class
4832                                         unsigned long order_idx = 0;
4833                                         while(( onode = jsonObjectGetIndex( snode, order_idx++ ) )) {
4834
4835                                                 const char* _f = jsonObjectGetString( onode );
4836
4837                                                 osrfHash* field_def = osrfHashGet( field_list_def, _f );
4838                                                 if( !field_def ) {
4839                                                         osrfLogError( OSRF_LOG_MARK,
4840                                                                         "%s: Invalid field \"%s\" in ORDER BY clause",
4841                                                                         modulename, _f );
4842                                                         if( ctx )
4843                                                                 osrfAppSessionStatus(
4844                                                                         ctx->session,
4845                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4846                                                                         "osrfMethodException",
4847                                                                         ctx->request,
4848                                                                         "Invalid field in ORDER BY clause -- "
4849                                                                         "see error log for more details"
4850                                                                 );
4851                                                         jsonIteratorFree( class_itr );
4852                                                         buffer_free( order_buf );
4853                                                         free(having_buf);
4854                                                         buffer_free(group_buf);
4855                                                         buffer_free(sql_buf);
4856                                                         if (defaultselhash)
4857                                                                 jsonObjectFree(defaultselhash);
4858                                                         return NULL;
4859                                                 } else if( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4860                                                         osrfLogError( OSRF_LOG_MARK,
4861                                                                 "%s: Virtual field \"%s\" in ORDER BY clause",
4862                                                                 modulename, _f );
4863                                                         if( ctx )
4864                                                                 osrfAppSessionStatus(
4865                                                                         ctx->session,
4866                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4867                                                                         "osrfMethodException",
4868                                                                         ctx->request,
4869                                                                         "Virtual field in ORDER BY clause -- "
4870                                                                         "see error log for more details"
4871                                                                 );
4872                                                         jsonIteratorFree( class_itr );
4873                                                         buffer_free( order_buf );
4874                                                         free(having_buf);
4875                                                         buffer_free(group_buf);
4876                                                         buffer_free(sql_buf);
4877                                                         if (defaultselhash)
4878                                                                 jsonObjectFree(defaultselhash);
4879                                                         return NULL;
4880                                                 }
4881
4882                                                 if ( order_buf )
4883                                                         OSRF_BUFFER_ADD(order_buf, ", ");
4884                                                 else
4885                                                         order_buf = buffer_init(128);
4886
4887                                                 buffer_fadd( order_buf, "\"%s\".%s", class_itr->key, _f);
4888
4889                                         } // end while
4890
4891                                 // IT'S THE OOOOOOOOOOOLD STYLE!
4892                                 } else {
4893                                         osrfLogError( OSRF_LOG_MARK,
4894                                                 "%s: Possible SQL injection attempt; direct order by is not allowed",
4895                                                 modulename );
4896                                         if (ctx) {
4897                                                 osrfAppSessionStatus(
4898                                                         ctx->session,
4899                                                         OSRF_STATUS_INTERNALSERVERERROR,
4900                                                         "osrfMethodException",
4901                                                         ctx->request,
4902                                                         "Severe query error -- see error log for more details"
4903                                                 );
4904                                         }
4905
4906                                         free(having_buf);
4907                                         buffer_free(group_buf);
4908                                         buffer_free(order_buf);
4909                                         buffer_free(sql_buf);
4910                                         if (defaultselhash)
4911                                                 jsonObjectFree(defaultselhash);
4912                                         jsonIteratorFree(class_itr);
4913                                         return NULL;
4914                                 }
4915                         } // end while
4916                         jsonIteratorFree( class_itr );
4917                 } else {
4918                         osrfLogError(OSRF_LOG_MARK,
4919                                 "%s: Malformed ORDER BY clause; expected JSON_HASH or JSON_ARRAY, found %s",
4920                                 modulename, json_type( order_hash->type ) );
4921                         if( ctx )
4922                                 osrfAppSessionStatus(
4923                                         ctx->session,
4924                                         OSRF_STATUS_INTERNALSERVERERROR,
4925                                         "osrfMethodException",
4926                                         ctx->request,
4927                                         "Malformed ORDER BY clause -- see error log for more details"
4928                                 );
4929                         buffer_free( order_buf );
4930                         free(having_buf);
4931                         buffer_free(group_buf);
4932                         buffer_free(sql_buf);
4933                         if (defaultselhash)
4934                                 jsonObjectFree(defaultselhash);
4935                         return NULL;
4936                 }
4937
4938                 if( order_buf )
4939                         order_by_list = buffer_release( order_buf );
4940         }
4941
4942
4943         string = buffer_release(group_buf);
4944
4945         if ( *string && ( aggregate_found || (flags & SELECT_DISTINCT) ) ) {
4946                 OSRF_BUFFER_ADD( sql_buf, " GROUP BY " );
4947                 OSRF_BUFFER_ADD( sql_buf, string );
4948         }
4949
4950         free(string);
4951
4952         if( having_buf && *having_buf ) {
4953                 OSRF_BUFFER_ADD( sql_buf, " HAVING " );
4954                 OSRF_BUFFER_ADD( sql_buf, having_buf );
4955                 free( having_buf );
4956         }
4957
4958         if( order_by_list ) {
4959
4960                 if ( *order_by_list ) {
4961                         OSRF_BUFFER_ADD( sql_buf, " ORDER BY " );
4962                         OSRF_BUFFER_ADD( sql_buf, order_by_list );
4963                 }
4964
4965                 free( order_by_list );
4966         }
4967
4968         if ( limit ){
4969                 const char* str = jsonObjectGetString(limit);
4970                 buffer_fadd( sql_buf, " LIMIT %d", atoi(str) );
4971         }
4972
4973         if (offset) {
4974                 const char* str = jsonObjectGetString(offset);
4975                 buffer_fadd( sql_buf, " OFFSET %d", atoi(str) );
4976         }
4977
4978         if (!(flags & SUBSELECT)) OSRF_BUFFER_ADD_CHAR(sql_buf, ';');
4979
4980         if (defaultselhash)
4981                  jsonObjectFree(defaultselhash);
4982
4983         return buffer_release(sql_buf);
4984
4985 } // end of SELECT()
4986
4987 static char* buildSELECT ( jsonObject* search_hash, jsonObject* order_hash, osrfHash* meta, osrfMethodContext* ctx ) {
4988
4989         const char* locale = osrf_message_get_last_locale();
4990
4991         osrfHash* fields = osrfHashGet(meta, "fields");
4992         char* core_class = osrfHashGet(meta, "classname");
4993
4994         const jsonObject* join_hash = jsonObjectGetKeyConst( order_hash, "join" );
4995
4996         jsonObject* node = NULL;
4997         jsonObject* snode = NULL;
4998         jsonObject* onode = NULL;
4999         const jsonObject* _tmp = NULL;
5000         jsonObject* selhash = NULL;
5001         jsonObject* defaultselhash = NULL;
5002
5003         growing_buffer* sql_buf = buffer_init(128);
5004         growing_buffer* select_buf = buffer_init(128);
5005
5006         if ( !(selhash = jsonObjectGetKey( order_hash, "select" )) ) {
5007                 defaultselhash = jsonNewObjectType(JSON_HASH);
5008                 selhash = defaultselhash;
5009         }
5010
5011         // If there's no SELECT list for the core class, build one
5012         if ( !jsonObjectGetKeyConst(selhash,core_class) ) {
5013                 jsonObject* field_list = jsonNewObjectType( JSON_ARRAY );
5014
5015                 // Add every non-virtual field to the field list
5016                 osrfHash* field_def = NULL;
5017                 osrfHashIterator* field_itr = osrfNewHashIterator( fields );
5018                 while( ( field_def = osrfHashIteratorNext( field_itr ) ) ) {
5019                         if( ! str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
5020                                 const char* field = osrfHashIteratorKey( field_itr );
5021                                 jsonObjectPush( field_list, jsonNewObject( field ) );
5022                         }
5023                 }
5024                 osrfHashIteratorFree( field_itr );
5025                 jsonObjectSetKey( selhash, core_class, field_list );
5026         }
5027
5028         int first = 1;
5029         jsonIterator* class_itr = jsonNewIterator( selhash );
5030         while ( (snode = jsonIteratorNext( class_itr )) ) {
5031
5032                 const char* cname = class_itr->key;
5033                 osrfHash* idlClass = osrfHashGet( oilsIDL(), cname );
5034                 if (!idlClass)
5035                         continue;
5036
5037                 if (strcmp(core_class,class_itr->key)) {
5038                         if (!join_hash)
5039                                 continue;
5040
5041                         jsonObject* found =  jsonObjectFindPath(join_hash, "//%s", class_itr->key);
5042                         if (!found->size) {
5043                                 jsonObjectFree(found);
5044                                 continue;
5045                         }
5046
5047                         jsonObjectFree(found);
5048                 }
5049
5050                 jsonIterator* select_itr = jsonNewIterator( snode );
5051                 while ( (node = jsonIteratorNext( select_itr )) ) {
5052                         const char* item_str = jsonObjectGetString( node );
5053                         osrfHash* field = osrfHashGet( osrfHashGet( idlClass, "fields" ), item_str );
5054                         char* fname = osrfHashGet(field, "name");
5055
5056                         if (!field)
5057                                 continue;
5058
5059                         if (first) {
5060                                 first = 0;
5061                         } else {
5062                                 OSRF_BUFFER_ADD_CHAR(select_buf, ',');
5063                         }
5064
5065                         if (locale) {
5066                                 const char* i18n;
5067                                 const jsonObject* no_i18n_obj = jsonObjectGetKey( order_hash, "no_i18n" );
5068                                 if ( obj_is_true( no_i18n_obj ) )    // Suppress internationalization?
5069                                         i18n = NULL;
5070                                 else
5071                                         i18n = osrfHashGet(field, "i18n");
5072
5073                                 if( str_is_true( i18n ) ) {
5074                                         char* pkey = osrfHashGet(idlClass, "primarykey");
5075                                         char* tname = osrfHashGet(idlClass, "tablename");
5076
5077                                         buffer_fadd(select_buf, " oils_i18n_xlate('%s', '%s', '%s', "
5078                                                         "'%s', \"%s\".%s::TEXT, '%s') AS \"%s\"",
5079                                                         tname, cname, fname, pkey, cname, pkey, locale, fname);
5080                                 } else {
5081                                         buffer_fadd(select_buf, " \"%s\".%s", cname, fname);
5082                                 }
5083                         } else {
5084                                 buffer_fadd(select_buf, " \"%s\".%s", cname, fname);
5085                         }
5086                 }
5087
5088                 jsonIteratorFree(select_itr);
5089         }
5090
5091         jsonIteratorFree(class_itr);
5092
5093         char* col_list = buffer_release(select_buf);
5094         char* table = getRelation(meta);
5095         if( !table )
5096                 table = strdup( "(null)" );
5097
5098         buffer_fadd(sql_buf, "SELECT %s FROM %s AS \"%s\"", col_list, table, core_class );
5099         free(col_list);
5100         free(table);
5101
5102         // Clear the query stack (as a fail-safe precaution against possible
5103         // leftover garbage); then push the first query frame onto the stack.
5104         clear_query_stack();
5105         push_query_frame();
5106         if( add_query_core( NULL, core_class ) ) {
5107                 if( ctx )
5108                         osrfAppSessionStatus(
5109                                 ctx->session,
5110                                 OSRF_STATUS_INTERNALSERVERERROR,
5111                                 "osrfMethodException",
5112                                 ctx->request,
5113                                 "Unable to build query frame for core class"
5114                         );
5115                 return NULL;
5116         }
5117
5118         if ( join_hash ) {
5119                 char* join_clause = searchJOIN( join_hash, &curr_query->core );
5120                 OSRF_BUFFER_ADD_CHAR(sql_buf, ' ');
5121                 OSRF_BUFFER_ADD(sql_buf, join_clause);
5122                 free(join_clause);
5123         }
5124
5125         osrfLogDebug( OSRF_LOG_MARK, "%s pre-predicate SQL =  %s",
5126                                   modulename, OSRF_BUFFER_C_STR( sql_buf ));
5127
5128         OSRF_BUFFER_ADD(sql_buf, " WHERE ");
5129
5130         char* pred = searchWHERE( search_hash, &curr_query->core, AND_OP_JOIN, ctx );
5131         if (!pred) {
5132                 osrfAppSessionStatus(
5133                         ctx->session,
5134                         OSRF_STATUS_INTERNALSERVERERROR,
5135                                 "osrfMethodException",
5136                                 ctx->request,
5137                                 "Severe query error -- see error log for more details"
5138                         );
5139                 buffer_free(sql_buf);
5140                 if(defaultselhash)
5141                         jsonObjectFree(defaultselhash);
5142                 clear_query_stack();
5143                 return NULL;
5144         } else {
5145                 buffer_add(sql_buf, pred);
5146                 free(pred);
5147         }
5148
5149         if (order_hash) {
5150                 char* string = NULL;
5151                 if ( (_tmp = jsonObjectGetKeyConst( order_hash, "order_by" )) ){
5152
5153                         growing_buffer* order_buf = buffer_init(128);
5154
5155                         first = 1;
5156                         jsonIterator* class_itr = jsonNewIterator( _tmp );
5157                         while ( (snode = jsonIteratorNext( class_itr )) ) {
5158
5159                                 if (!jsonObjectGetKeyConst(selhash,class_itr->key))
5160                                         continue;
5161
5162                                 if ( snode->type == JSON_HASH ) {
5163
5164                                         jsonIterator* order_itr = jsonNewIterator( snode );
5165                                         while ( (onode = jsonIteratorNext( order_itr )) ) {
5166
5167                                                 osrfHash* field_def = oilsIDLFindPath( "/%s/fields/%s",
5168                                                                 class_itr->key, order_itr->key );
5169                                                 if ( !field_def )
5170                                                         continue;
5171
5172                                                 char* direction = NULL;
5173                                                 if ( onode->type == JSON_HASH ) {
5174                                                         if ( jsonObjectGetKeyConst( onode, "transform" ) ) {
5175                                                                 string = searchFieldTransform( class_itr->key, field_def, onode );
5176                                                                 if( ! string ) {
5177                                                                         osrfAppSessionStatus(
5178                                                                                 ctx->session,
5179                                                                                 OSRF_STATUS_INTERNALSERVERERROR,
5180                                                                                 "osrfMethodException",
5181                                                                                 ctx->request,
5182                                                                                 "Severe query error in ORDER BY clause -- "
5183                                                                                 "see error log for more details"
5184                                                                         );
5185                                                                         jsonIteratorFree( order_itr );
5186                                                                         jsonIteratorFree( class_itr );
5187                                                                         buffer_free( order_buf );
5188                                                                         buffer_free( sql_buf );
5189                                                                         if( defaultselhash )
5190                                                                                 jsonObjectFree( defaultselhash );
5191                                                                         clear_query_stack();
5192                                                                         return NULL;
5193                                                                 }
5194                                                         } else {
5195                                                                 growing_buffer* field_buf = buffer_init(16);
5196                                                                 buffer_fadd( field_buf, "\"%s\".%s",
5197                                                                         class_itr->key, order_itr->key );
5198                                                                 string = buffer_release(field_buf);
5199                                                         }
5200
5201                                                         if ( (_tmp = jsonObjectGetKeyConst( onode, "direction" )) ) {
5202                                                                 const char* dir = jsonObjectGetString(_tmp);
5203                                                                 if (!strncasecmp(dir, "d", 1)) {
5204                                                                         direction = " DESC";
5205                                                                 } else {
5206                                                                         free(direction);
5207                                                                 }
5208                                                         }
5209                                                 } else {
5210                                                         string = strdup(order_itr->key);
5211                                                         const char* dir = jsonObjectGetString(onode);
5212                                                         if (!strncasecmp(dir, "d", 1)) {
5213                                                                 direction = " DESC";
5214                                                         } else {
5215                                                                 direction = " ASC";
5216                                                         }
5217                                                 }
5218
5219                                                 if (first) {
5220                                                         first = 0;
5221                                                 } else {
5222                                                         buffer_add(order_buf, ", ");
5223                                                 }
5224
5225                                                 buffer_add(order_buf, string);
5226                                                 free(string);
5227
5228                                                 if (direction) {
5229                                                         buffer_add(order_buf, direction);
5230                                                 }
5231                                         }
5232
5233                                         jsonIteratorFree(order_itr);
5234
5235                                 } else {
5236                                         const char* str = jsonObjectGetString(snode);
5237                                         buffer_add(order_buf, str);
5238                                         break;
5239                                 }
5240
5241                         }
5242
5243                         jsonIteratorFree(class_itr);
5244
5245                         string = buffer_release(order_buf);
5246
5247                         if ( *string ) {
5248                                 OSRF_BUFFER_ADD( sql_buf, " ORDER BY " );
5249                                 OSRF_BUFFER_ADD( sql_buf, string );
5250                         }
5251
5252                         free(string);
5253                 }
5254
5255                 if ( (_tmp = jsonObjectGetKeyConst( order_hash, "limit" )) ){
5256                         const char* str = jsonObjectGetString(_tmp);
5257                         buffer_fadd(
5258                                 sql_buf,
5259                                 " LIMIT %d",
5260                                 atoi(str)
5261                         );
5262                 }
5263
5264                 _tmp = jsonObjectGetKeyConst( order_hash, "offset" );
5265                 if (_tmp) {
5266                         const char* str = jsonObjectGetString(_tmp);
5267                         buffer_fadd(
5268                                 sql_buf,
5269                                 " OFFSET %d",
5270                                 atoi(str)
5271                         );
5272                 }
5273         }
5274
5275         if (defaultselhash)
5276                 jsonObjectFree(defaultselhash);
5277         clear_query_stack();
5278
5279         OSRF_BUFFER_ADD_CHAR(sql_buf, ';');
5280         return buffer_release(sql_buf);
5281 }
5282
5283 int doJSONSearch ( osrfMethodContext* ctx ) {
5284         if(osrfMethodVerifyContext( ctx )) {
5285                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
5286                 return -1;
5287         }
5288
5289         osrfLogDebug(OSRF_LOG_MARK, "Received query request");
5290
5291         int err = 0;
5292
5293         // XXX for now...
5294         dbhandle = writehandle;
5295
5296         jsonObject* hash = jsonObjectGetIndex(ctx->params, 0);
5297
5298         int flags = 0;
5299
5300         if ( obj_is_true( jsonObjectGetKey( hash, "distinct" ) ) )
5301                 flags |= SELECT_DISTINCT;
5302
5303         if ( obj_is_true( jsonObjectGetKey( hash, "no_i18n" ) ) )
5304                 flags |= DISABLE_I18N;
5305
5306         osrfLogDebug(OSRF_LOG_MARK, "Building SQL ...");
5307         clear_query_stack();       // a possibly needless precaution
5308         char* sql = buildQuery( ctx, hash, flags );
5309         clear_query_stack();
5310
5311         if (!sql) {
5312                 err = -1;
5313                 return err;
5314         }
5315
5316         osrfLogDebug(OSRF_LOG_MARK, "%s SQL =  %s", modulename, sql);
5317         dbi_result result = dbi_conn_query(dbhandle, sql);
5318
5319         if(result) {
5320                 osrfLogDebug(OSRF_LOG_MARK, "Query returned with no errors");
5321
5322                 if (dbi_result_first_row(result)) {
5323                         /* JSONify the result */
5324                         osrfLogDebug(OSRF_LOG_MARK, "Query returned at least one row");
5325
5326                         do {
5327                                 jsonObject* return_val = oilsMakeJSONFromResult( result );
5328                                 osrfAppRespond( ctx, return_val );
5329                                 jsonObjectFree( return_val );
5330                         } while (dbi_result_next_row(result));
5331
5332                 } else {
5333                         osrfLogDebug(OSRF_LOG_MARK, "%s returned no results for query %s", modulename, sql);
5334                 }
5335
5336                 osrfAppRespondComplete( ctx, NULL );
5337
5338                 /* clean up the query */
5339                 dbi_result_free(result);
5340
5341         } else {
5342                 err = -1;
5343                 osrfLogError(OSRF_LOG_MARK, "%s: Error with query [%s]", modulename, sql);
5344                 osrfAppSessionStatus(
5345                         ctx->session,
5346                         OSRF_STATUS_INTERNALSERVERERROR,
5347                         "osrfMethodException",
5348                         ctx->request,
5349                         "Severe query error -- see error log for more details"
5350                 );
5351         }
5352
5353         free(sql);
5354         return err;
5355 }
5356
5357 // The last parameter, err, is used to report an error condition by updating an int owned by
5358 // the calling code.
5359
5360 // In case of an error, we set *err to -1.  If there is no error, *err is left unchanged.
5361 // It is the responsibility of the calling code to initialize *err before the
5362 // call, so that it will be able to make sense of the result.
5363
5364 // Note also that we return NULL if and only if we set *err to -1.  So the err parameter is
5365 // redundant anyway.
5366 static jsonObject* doFieldmapperSearch ( osrfMethodContext* ctx, osrfHash* class_meta,
5367                 jsonObject* where_hash, jsonObject* query_hash, int* err ) {
5368
5369         // XXX for now...
5370         dbhandle = writehandle;
5371
5372         char* core_class = osrfHashGet( class_meta, "classname" );
5373         char* pkey = osrfHashGet( class_meta, "primarykey" );
5374
5375         const jsonObject* _tmp;
5376
5377         char* sql = buildSELECT( where_hash, query_hash, class_meta, ctx );
5378         if (!sql) {
5379                 osrfLogDebug(OSRF_LOG_MARK, "Problem building query, returning NULL");
5380                 *err = -1;
5381                 return NULL;
5382         }
5383
5384         osrfLogDebug(OSRF_LOG_MARK, "%s SQL =  %s", modulename, sql);
5385
5386         dbi_result result = dbi_conn_query(dbhandle, sql);
5387         if( NULL == result ) {
5388                 osrfLogError(OSRF_LOG_MARK, "%s: Error retrieving %s with query [%s]",
5389                         modulename, osrfHashGet( class_meta, "fieldmapper" ), sql);
5390                 osrfAppSessionStatus(
5391                         ctx->session,
5392                         OSRF_STATUS_INTERNALSERVERERROR,
5393                         "osrfMethodException",
5394                         ctx->request,
5395                         "Severe query error -- see error log for more details"
5396                 );
5397                 *err = -1;
5398                 free(sql);
5399                 return jsonNULL;
5400
5401         } else {
5402                 osrfLogDebug(OSRF_LOG_MARK, "Query returned with no errors");
5403         }
5404
5405         jsonObject* res_list = jsonNewObjectType(JSON_ARRAY);
5406         jsonObject* row_obj = NULL;
5407
5408         if (dbi_result_first_row(result)) {
5409
5410                 // Convert each row to a JSON_ARRAY of column values, and enclose those objects
5411                 // in a JSON_ARRAY of rows.  If two or more rows have the same key value, then
5412                 // eliminate the duplicates.
5413                 osrfLogDebug(OSRF_LOG_MARK, "Query returned at least one row");
5414                 osrfHash* dedup = osrfNewHash();
5415                 do {
5416                         row_obj = oilsMakeFieldmapperFromResult( result, class_meta );
5417                         char* pkey_val = oilsFMGetString( row_obj, pkey );
5418                         if ( osrfHashGet( dedup, pkey_val ) ) {
5419                                 jsonObjectFree( row_obj );
5420                                 free( pkey_val );
5421                         } else {
5422                                 osrfHashSet( dedup, pkey_val, pkey_val );
5423                                 jsonObjectPush( res_list, row_obj );
5424                         }
5425                 } while (dbi_result_next_row(result));
5426                 osrfHashFree(dedup);
5427
5428         } else {
5429                 osrfLogDebug(OSRF_LOG_MARK, "%s returned no results for query %s",
5430                         modulename, sql );
5431         }
5432
5433         /* clean up the query */
5434         dbi_result_free(result);
5435         free(sql);
5436
5437         // If we're asked to flesh, and there's anything to flesh, then flesh.
5438         if (res_list->size && query_hash) {
5439                 _tmp = jsonObjectGetKeyConst( query_hash, "flesh" );
5440                 if (_tmp) {
5441                         // Get the flesh depth
5442                         int flesh_depth = (int) jsonObjectGetNumber( _tmp );
5443                         if ( flesh_depth == -1 || flesh_depth > max_flesh_depth )
5444                                 flesh_depth = max_flesh_depth;
5445
5446                         // We need a non-zero flesh depth, and a list of fields to flesh
5447                         const jsonObject* temp_blob = jsonObjectGetKeyConst( query_hash, "flesh_fields" );
5448                         if ( temp_blob && flesh_depth > 0 ) {
5449
5450                                 jsonObject* flesh_blob = jsonObjectClone( temp_blob );
5451                                 const jsonObject* flesh_fields = jsonObjectGetKeyConst( flesh_blob, core_class );
5452
5453                                 osrfStringArray* link_fields = NULL;
5454                                 osrfHash* links = osrfHashGet( class_meta, "links" );
5455
5456                                 // Make an osrfStringArray of the names of fields to be fleshed
5457                                 if (flesh_fields) {
5458                                         if (flesh_fields->size == 1) {
5459                                                 const char* _t = jsonObjectGetString(
5460                                                         jsonObjectGetIndex( flesh_fields, 0 ) );
5461                                                 if (!strcmp(_t,"*"))
5462                                                         link_fields = osrfHashKeys( links );
5463                                         }
5464
5465                                         if (!link_fields) {
5466                                                 jsonObject* _f;
5467                                                 link_fields = osrfNewStringArray(1);
5468                                                 jsonIterator* _i = jsonNewIterator( flesh_fields );
5469                                                 while ((_f = jsonIteratorNext( _i ))) {
5470                                                         osrfStringArrayAdd( link_fields, jsonObjectGetString( _f ) );
5471                                                 }
5472                                                 jsonIteratorFree(_i);
5473                                         }
5474                                 }
5475
5476                                 osrfHash* fields = osrfHashGet( class_meta, "fields" );
5477
5478                                 // Iterate over the JSON_ARRAY of rows
5479                                 jsonObject* cur;
5480                                 unsigned long res_idx = 0;
5481                                 while ((cur = jsonObjectGetIndex( res_list, res_idx++ ) )) {
5482
5483                                         int i = 0;
5484                                         const char* link_field;
5485
5486                                         // Iterate over the list of fleshable fields
5487                                         while ( (link_field = osrfStringArrayGetString(link_fields, i++)) ) {
5488
5489                                                 osrfLogDebug(OSRF_LOG_MARK, "Starting to flesh %s", link_field);
5490
5491                                                 osrfHash* kid_link = osrfHashGet(links, link_field);
5492                                                 if (!kid_link)
5493                                                         continue;     // Not a link field; skip it
5494
5495                                                 osrfHash* field = osrfHashGet(fields, link_field);
5496                                                 if (!field)
5497                                                         continue;     // Not a field at all; skip it (IDL is ill-formed)
5498
5499                                                 osrfHash* kid_idl = osrfHashGet(oilsIDL(), osrfHashGet(kid_link, "class"));
5500                                                 if (!kid_idl)
5501                                                         continue;   // The class it links to doesn't exist; skip it
5502
5503                                                 const char* reltype = osrfHashGet( kid_link, "reltype" );
5504                                                 if( !reltype )
5505                                                         continue;   // No reltype; skip it (IDL is ill-formed)
5506
5507                                                 osrfHash* value_field = field;
5508
5509                                                 if (   !strcmp( reltype, "has_many" )
5510                                                         || !strcmp( reltype, "might_have" ) ) { // has_many or might_have
5511                                                         value_field = osrfHashGet(
5512                                                                 fields, osrfHashGet( class_meta, "primarykey" ) );
5513                                                 }
5514
5515                                                 osrfStringArray* link_map = osrfHashGet( kid_link, "map" );
5516
5517                                                 if (link_map->size > 0) {
5518                                                         jsonObject* _kid_key = jsonNewObjectType(JSON_ARRAY);
5519                                                         jsonObjectPush(
5520                                                                 _kid_key,
5521                                                                 jsonNewObject( osrfStringArrayGetString( link_map, 0 ) )
5522                                                         );
5523
5524                                                         jsonObjectSetKey(
5525                                                                 flesh_blob,
5526                                                                 osrfHashGet(kid_link, "class"),
5527                                                                 _kid_key
5528                                                         );
5529                                                 };
5530
5531                                                 osrfLogDebug(
5532                                                         OSRF_LOG_MARK,
5533                                                         "Link field: %s, remote class: %s, fkey: %s, reltype: %s",
5534                                                         osrfHashGet(kid_link, "field"),
5535                                                         osrfHashGet(kid_link, "class"),
5536                                                         osrfHashGet(kid_link, "key"),
5537                                                         osrfHashGet(kid_link, "reltype")
5538                                                 );
5539
5540                                                 const char* search_key = jsonObjectGetString(
5541                                                         jsonObjectGetIndex( cur,
5542                                                                 atoi( osrfHashGet(value_field, "array_position") )
5543                                                         )
5544                                                 );
5545
5546                                                 if (!search_key) {
5547                                                         osrfLogDebug(OSRF_LOG_MARK, "Nothing to search for!");
5548                                                         continue;
5549                                                 }
5550
5551                                                 osrfLogDebug(OSRF_LOG_MARK, "Creating param objects...");
5552
5553                                                 // construct WHERE clause
5554                                                 jsonObject* where_clause  = jsonNewObjectType(JSON_HASH);
5555                                                 jsonObjectSetKey(
5556                                                         where_clause,
5557                                                         osrfHashGet(kid_link, "key"),
5558                                                         jsonNewObject( search_key )
5559                                                 );
5560
5561                                                 // construct the rest of the query, mostly
5562                                                 // by copying pieces of the previous level of query
5563                                                 jsonObject* rest_of_query = jsonNewObjectType(JSON_HASH);
5564                                                 jsonObjectSetKey( rest_of_query, "flesh",
5565                                                         jsonNewNumberObject( flesh_depth - 1 + link_map->size )
5566                                                 );
5567
5568                                                 if (flesh_blob)
5569                                                         jsonObjectSetKey( rest_of_query, "flesh_fields",
5570                                                                 jsonObjectClone(flesh_blob) );
5571
5572                                                 if (jsonObjectGetKeyConst(query_hash, "order_by")) {
5573                                                         jsonObjectSetKey( rest_of_query, "order_by",
5574                                                                 jsonObjectClone(jsonObjectGetKeyConst(query_hash, "order_by"))
5575                                                         );
5576                                                 }
5577
5578                                                 if (jsonObjectGetKeyConst(query_hash, "select")) {
5579                                                         jsonObjectSetKey( rest_of_query, "select",
5580                                                                 jsonObjectClone(jsonObjectGetKeyConst(query_hash, "select"))
5581                                                         );
5582                                                 }
5583
5584                                                 // do the query, recursively, to expand the fleshable field
5585                                                 jsonObject* kids = doFieldmapperSearch( ctx, kid_idl,
5586                                                         where_clause, rest_of_query, err);
5587
5588                                                 jsonObjectFree( where_clause );
5589                                                 jsonObjectFree( rest_of_query );
5590
5591                                                 if(*err) {
5592                                                         osrfStringArrayFree(link_fields);
5593                                                         jsonObjectFree(res_list);
5594                                                         jsonObjectFree(flesh_blob);
5595                                                         return jsonNULL;
5596                                                 }
5597
5598                                                 osrfLogDebug(OSRF_LOG_MARK, "Search for %s return %d linked objects",
5599                                                         osrfHashGet(kid_link, "class"), kids->size);
5600
5601                                                 // Traverse the result set
5602                                                 jsonObject* X = NULL;
5603                                                 if ( link_map->size > 0 && kids->size > 0 ) {
5604                                                         X = kids;
5605                                                         kids = jsonNewObjectType(JSON_ARRAY);
5606
5607                                                         jsonObject* _k_node;
5608                                                         unsigned long res_idx = 0;
5609                                                         while ((_k_node = jsonObjectGetIndex( X, res_idx++ ) )) {
5610                                                                 jsonObjectPush(
5611                                                                         kids,
5612                                                                         jsonObjectClone(
5613                                                                                 jsonObjectGetIndex(
5614                                                                                         _k_node,
5615                                                                                         (unsigned long)atoi(
5616                                                                                                 osrfHashGet(
5617                                                                                                         osrfHashGet(
5618                                                                                                                 osrfHashGet(
5619                                                                                                                         osrfHashGet(
5620                                                                                                                                 oilsIDL(),
5621                                                                                                                                 osrfHashGet(kid_link, "class")
5622                                                                                                                         ),
5623                                                                                                                         "fields"
5624                                                                                                                 ),
5625                                                                                                                 osrfStringArrayGetString( link_map, 0 )
5626                                                                                                         ),
5627                                                                                                         "array_position"
5628                                                                                                 )
5629                                                                                         )
5630                                                                                 )
5631                                                                         )
5632                                                                 );
5633                                                         } // end while loop traversing X
5634                                                 }
5635
5636                                                 if (!(strcmp( osrfHashGet(kid_link, "reltype"), "has_a" ))
5637                                                         || !(strcmp( osrfHashGet(kid_link, "reltype"), "might_have" ))) {
5638                                                         osrfLogDebug(OSRF_LOG_MARK, "Storing fleshed objects in %s",
5639                                                                 osrfHashGet(kid_link, "field"));
5640                                                         jsonObjectSetIndex(
5641                                                                 cur,
5642                                                                 (unsigned long)atoi( osrfHashGet( field, "array_position" ) ),
5643                                                                 jsonObjectClone( jsonObjectGetIndex(kids, 0) )
5644                                                         );
5645                                                 }
5646
5647                                                 if (!(strcmp( osrfHashGet(kid_link, "reltype"), "has_many" ))) {
5648                                                         // has_many
5649                                                         osrfLogDebug( OSRF_LOG_MARK, "Storing fleshed objects in %s",
5650                                                                 osrfHashGet( kid_link, "field" ) );
5651                                                         jsonObjectSetIndex(
5652                                                                 cur,
5653                                                                 (unsigned long)atoi( osrfHashGet( field, "array_position" ) ),
5654                                                                 jsonObjectClone( kids )
5655                                                         );
5656                                                 }
5657
5658                                                 if (X) {
5659                                                         jsonObjectFree(kids);
5660                                                         kids = X;
5661                                                 }
5662
5663                                                 jsonObjectFree( kids );
5664
5665                                                 osrfLogDebug( OSRF_LOG_MARK, "Fleshing of %s complete",
5666                                                         osrfHashGet( kid_link, "field" ) );
5667                                                 osrfLogDebug(OSRF_LOG_MARK, "%s", jsonObjectToJSON(cur));
5668
5669                                         } // end while loop traversing list of fleshable fields
5670                                 } // end while loop traversing res_list
5671                                 jsonObjectFree( flesh_blob );
5672                                 osrfStringArrayFree(link_fields);
5673                         }
5674                 }
5675         }
5676
5677         return res_list;
5678 }
5679
5680
5681 static int doUpdate(osrfMethodContext* ctx ) {
5682         if(osrfMethodVerifyContext( ctx )) {
5683                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
5684                 return -1;
5685         }
5686
5687         if( enforce_pcrud )
5688                 timeout_needs_resetting = 1;
5689
5690         osrfHash* meta = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );
5691
5692         jsonObject* target = NULL;
5693         if( enforce_pcrud )
5694                 target = jsonObjectGetIndex( ctx->params, 1 );
5695         else
5696                 target = jsonObjectGetIndex( ctx->params, 0 );
5697
5698         if (!verifyObjectClass(ctx, target)) {
5699                 osrfAppRespondComplete( ctx, NULL );
5700                 return -1;
5701         }
5702
5703         if( getXactId( ctx ) == NULL ) {
5704                 osrfAppSessionStatus(
5705                         ctx->session,
5706                         OSRF_STATUS_BADREQUEST,
5707                         "osrfMethodException",
5708                         ctx->request,
5709                         "No active transaction -- required for UPDATE"
5710                 );
5711                 osrfAppRespondComplete( ctx, NULL );
5712                 return -1;
5713         }
5714
5715         // The following test is harmless but redundant.  If a class is
5716         // readonly, we don't register an update method for it.
5717         if( str_is_true( osrfHashGet( meta, "readonly" ) ) ) {
5718                 osrfAppSessionStatus(
5719                         ctx->session,
5720                         OSRF_STATUS_BADREQUEST,
5721                         "osrfMethodException",
5722                         ctx->request,
5723                         "Cannot UPDATE readonly class"
5724                 );
5725                 osrfAppRespondComplete( ctx, NULL );
5726                 return -1;
5727         }
5728
5729         dbhandle = writehandle;
5730         const char* trans_id = getXactId( ctx );
5731
5732         // Set the last_xact_id
5733         int index = oilsIDL_ntop( target->classname, "last_xact_id" );
5734         if (index > -1) {
5735                 osrfLogDebug(OSRF_LOG_MARK, "Setting last_xact_id to %s on %s at position %d",
5736                                 trans_id, target->classname, index);
5737                 jsonObjectSetIndex(target, index, jsonNewObject(trans_id));
5738         }
5739
5740         char* pkey = osrfHashGet(meta, "primarykey");
5741         osrfHash* fields = osrfHashGet(meta, "fields");
5742
5743         char* id = oilsFMGetString( target, pkey );
5744
5745         osrfLogDebug(
5746                 OSRF_LOG_MARK,
5747                 "%s updating %s object with %s = %s",
5748                 modulename,
5749                 osrfHashGet(meta, "fieldmapper"),
5750                 pkey,
5751                 id
5752         );
5753
5754         growing_buffer* sql = buffer_init(128);
5755         buffer_fadd(sql,"UPDATE %s SET", osrfHashGet(meta, "tablename"));
5756
5757         int first = 1;
5758         osrfHash* field_def = NULL;
5759         osrfHashIterator* field_itr = osrfNewHashIterator( fields );
5760         while( (field_def = osrfHashIteratorNext( field_itr ) ) ) {
5761
5762                 // Skip virtual fields, and the primary key
5763                 if( str_is_true( osrfHashGet( field_def, "virtual") ) )
5764                         continue;
5765
5766                 const char* field_name = osrfHashIteratorKey( field_itr );
5767                 if( ! strcmp( field_name, pkey ) )
5768                         continue;
5769
5770                 const jsonObject* field_object = oilsFMGetObject( target, field_name );
5771
5772                 int value_is_numeric = 0;    // boolean
5773                 char* value;
5774                 if (field_object && field_object->classname) {
5775                         value = oilsFMGetString(
5776                                 field_object,
5777                                 (char*)oilsIDLFindPath("/%s/primarykey", field_object->classname)
5778                         );
5779                 } else if( field_object && JSON_BOOL == field_object->type ) {
5780                         if( jsonBoolIsTrue( field_object ) )
5781                                 value = strdup( "t" );
5782                         else
5783                                 value = strdup( "f" );
5784                 } else {
5785                         value = jsonObjectToSimpleString( field_object );
5786                         if( field_object && JSON_NUMBER == field_object->type )
5787                                 value_is_numeric = 1;
5788                 }
5789
5790                 osrfLogDebug( OSRF_LOG_MARK, "Updating %s object with %s = %s",
5791                                 osrfHashGet(meta, "fieldmapper"), field_name, value);
5792
5793                 if (!field_object || field_object->type == JSON_NULL) {
5794                         if ( !(!( strcmp( osrfHashGet(meta, "classname"), "au" ) )
5795                                         && !( strcmp( field_name, "passwd" ) )) ) { // arg at the special case!
5796                                 if( first )
5797                                         first = 0;
5798                                 else
5799                                         OSRF_BUFFER_ADD_CHAR(sql, ',');
5800                                 buffer_fadd( sql, " %s = NULL", field_name );
5801                         }
5802
5803                 } else if ( value_is_numeric || !strcmp( get_primitive( field_def ), "number") ) {
5804                         if( first )
5805                                 first = 0;
5806                         else
5807                                 OSRF_BUFFER_ADD_CHAR(sql, ',');
5808
5809                         const char* numtype = get_datatype( field_def );
5810                         if ( !strncmp( numtype, "INT", 3 ) ) {
5811                                 buffer_fadd( sql, " %s = %ld", field_name, atol(value) );
5812                         } else if ( !strcmp( numtype, "NUMERIC" ) ) {
5813                                 buffer_fadd( sql, " %s = %f", field_name, atof(value) );
5814                         } else {
5815                                 // Must really be intended as a string, so quote it
5816                                 if ( dbi_conn_quote_string(dbhandle, &value) ) {
5817                                         buffer_fadd( sql, " %s = %s", field_name, value );
5818                                 } else {
5819                                         osrfLogError(OSRF_LOG_MARK, "%s: Error quoting string [%s]",
5820                                                 modulename, value);
5821                                         osrfAppSessionStatus(
5822                                                 ctx->session,
5823                                                 OSRF_STATUS_INTERNALSERVERERROR,
5824                                                 "osrfMethodException",
5825                                                 ctx->request,
5826                                                 "Error quoting string -- please see the error log for more details"
5827                                         );
5828                                         free(value);
5829                                         free(id);
5830                                         osrfHashIteratorFree( field_itr );
5831                                         buffer_free(sql);
5832                                         osrfAppRespondComplete( ctx, NULL );
5833                                         return -1;
5834                                 }
5835                         }
5836
5837                         osrfLogDebug( OSRF_LOG_MARK, "%s is of type %s", field_name, numtype );
5838
5839                 } else {
5840                         if ( dbi_conn_quote_string(dbhandle, &value) ) {
5841                                 if(first)
5842                                         first = 0;
5843                                 else
5844                                         OSRF_BUFFER_ADD_CHAR( sql, ',' );
5845                                 buffer_fadd( sql, " %s = %s", field_name, value );
5846                         } else {
5847                                 osrfLogError(OSRF_LOG_MARK, "%s: Error quoting string [%s]", modulename, value);
5848                                 osrfAppSessionStatus(
5849                                         ctx->session,
5850                                         OSRF_STATUS_INTERNALSERVERERROR,
5851                                         "osrfMethodException",
5852                                         ctx->request,
5853                                         "Error quoting string -- please see the error log for more details"
5854                                 );
5855                                 free(value);
5856                                 free(id);
5857                                 osrfHashIteratorFree( field_itr );
5858                                 buffer_free(sql);
5859                                 osrfAppRespondComplete( ctx, NULL );
5860                                 return -1;
5861                         }
5862                 }
5863
5864                 free(value);
5865
5866         } // end while
5867
5868         osrfHashIteratorFree( field_itr );
5869
5870         jsonObject* obj = jsonNewObject(id);
5871
5872         if ( strcmp( get_primitive( osrfHashGet( osrfHashGet(meta, "fields"), pkey ) ), "number" ) )
5873                 dbi_conn_quote_string(dbhandle, &id);
5874
5875         buffer_fadd( sql, " WHERE %s = %s;", pkey, id );
5876
5877         char* query = buffer_release(sql);
5878         osrfLogDebug(OSRF_LOG_MARK, "%s: Update SQL [%s]", modulename, query);
5879
5880         dbi_result result = dbi_conn_query(dbhandle, query);
5881         free(query);
5882
5883         if (!result) {
5884                 jsonObjectFree(obj);
5885                 obj = jsonNewObject(NULL);
5886                 osrfLogError(
5887                         OSRF_LOG_MARK,
5888                         "%s ERROR updating %s object with %s = %s",
5889                         modulename,
5890                         osrfHashGet(meta, "fieldmapper"),
5891                         pkey,
5892                         id
5893                 );
5894         }
5895
5896         free(id);
5897         osrfAppRespondComplete( ctx, obj );
5898         jsonObjectFree( obj );
5899         return 0;
5900 }
5901
5902 static int doDelete( osrfMethodContext* ctx ) {
5903         if(osrfMethodVerifyContext( ctx )) {
5904                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
5905                 return -1;
5906         }
5907
5908         if( enforce_pcrud )
5909                 timeout_needs_resetting = 1;
5910
5911         osrfHash* meta = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );
5912
5913         if( getXactId( ctx ) == NULL ) {
5914                 osrfAppSessionStatus(
5915                         ctx->session,
5916                         OSRF_STATUS_BADREQUEST,
5917                         "osrfMethodException",
5918                         ctx->request,
5919                         "No active transaction -- required for DELETE"
5920                 );
5921                 osrfAppRespondComplete( ctx, NULL );
5922                 return -1;
5923         }
5924
5925         // The following test is harmless but redundant.  If a class is
5926         // readonly, we don't register a delete method for it.
5927         if( str_is_true( osrfHashGet( meta, "readonly" ) ) ) {
5928                 osrfAppSessionStatus(
5929                         ctx->session,
5930                         OSRF_STATUS_BADREQUEST,
5931                         "osrfMethodException",
5932                         ctx->request,
5933                         "Cannot DELETE readonly class"
5934                 );
5935                 osrfAppRespondComplete( ctx, NULL );
5936                 return -1;
5937         }
5938
5939         dbhandle = writehandle;
5940
5941         char* pkey = osrfHashGet(meta, "primarykey");
5942
5943         int _obj_pos = 0;
5944         if( enforce_pcrud )
5945                 _obj_pos = 1;
5946
5947         char* id;
5948         if (jsonObjectGetIndex(ctx->params, _obj_pos)->classname) {
5949                 if (!verifyObjectClass(ctx, jsonObjectGetIndex( ctx->params, _obj_pos ))) {
5950                         osrfAppRespondComplete( ctx, NULL );
5951                         return -1;
5952                 }
5953
5954                 id = oilsFMGetString( jsonObjectGetIndex(ctx->params, _obj_pos), pkey );
5955         } else {
5956                 if( enforce_pcrud && !verifyObjectPCRUD( ctx, NULL )) {
5957                         osrfAppRespondComplete( ctx, NULL );
5958                         return -1;
5959                 }
5960                 id = jsonObjectToSimpleString(jsonObjectGetIndex(ctx->params, _obj_pos));
5961         }
5962
5963         osrfLogDebug(
5964                 OSRF_LOG_MARK,
5965                 "%s deleting %s object with %s = %s",
5966                 modulename,
5967                 osrfHashGet(meta, "fieldmapper"),
5968                 pkey,
5969                 id
5970         );
5971
5972         jsonObject* obj = jsonNewObject(id);
5973
5974         if ( strcmp( get_primitive( osrfHashGet( osrfHashGet(meta, "fields"), pkey ) ), "number" ) )
5975                 dbi_conn_quote_string(writehandle, &id);
5976
5977         dbi_result result = dbi_conn_queryf(writehandle, "DELETE FROM %s WHERE %s = %s;",
5978                 osrfHashGet(meta, "tablename"), pkey, id);
5979
5980         if (!result) {
5981                 jsonObjectFree(obj);
5982                 obj = jsonNewObject(NULL);
5983                 osrfLogError(
5984                         OSRF_LOG_MARK,
5985                         "%s ERROR deleting %s object with %s = %s",
5986                         modulename,
5987                         osrfHashGet(meta, "fieldmapper"),
5988                         pkey,
5989                         id
5990                 );
5991         }
5992
5993         free(id);
5994
5995         osrfAppRespondComplete( ctx, obj );
5996         jsonObjectFree( obj );
5997         return 0;
5998 }
5999
6000 /**
6001         @brief Translate a row returned from the database into a jsonObject of type JSON_ARRAY.
6002         @param result An iterator for a result set; we only look at the current row.
6003         @param @meta Pointer to the class metadata for the core class.
6004         @return Pointer to the resulting jsonObject if successful; otherwise NULL.
6005
6006         If a column is not defined in the IDL, or if it has no array_position defined for it in
6007         the IDL, or if it is defined as virtual, ignore it.
6008
6009         Otherwise, translate the column value into a jsonObject of type JSON_NULL, JSON_NUMBER,
6010         or JSON_STRING.  Then insert this jsonObject into the JSON_ARRAY according to its
6011         array_position in the IDL.
6012
6013         A field defined in the IDL but not represented in the returned row will leave a hole
6014         in the JSON_ARRAY.  In effect it will be treated as a null value.
6015
6016         In the resulting JSON_ARRAY, the field values appear in the sequence defined by the IDL,
6017         regardless of their sequence in the SELECT statement.  The JSON_ARRAY is assigned the
6018         classname corresponding to the @a meta argument.
6019
6020         The calling code is responsible for freeing the the resulting jsonObject by calling
6021         jsonObjectFree().
6022 */
6023 static jsonObject* oilsMakeFieldmapperFromResult( dbi_result result, osrfHash* meta) {
6024         if(!(result && meta)) return jsonNULL;
6025
6026         jsonObject* object = jsonNewObjectType( JSON_ARRAY );
6027         jsonObjectSetClass(object, osrfHashGet(meta, "classname"));
6028         osrfLogInternal(OSRF_LOG_MARK, "Setting object class to %s ", object->classname);
6029
6030         osrfHash* fields = osrfHashGet(meta, "fields");
6031
6032         int columnIndex = 1;
6033         const char* columnName;
6034
6035         /* cycle through the columns in the row returned from the database */
6036         while( (columnName = dbi_result_get_field_name(result, columnIndex)) ) {
6037
6038                 osrfLogInternal(OSRF_LOG_MARK, "Looking for column named [%s]...", (char*)columnName);
6039
6040                 int fmIndex = -1;  // Will be set to the IDL's sequence number for this field
6041
6042                 /* determine the field type and storage attributes */
6043                 unsigned short type = dbi_result_get_field_type_idx(result, columnIndex);
6044                 int attr            = dbi_result_get_field_attribs_idx(result, columnIndex);
6045
6046                 // Fetch the IDL's sequence number for the field.  If the field isn't in the IDL,
6047                 // or if it has no sequence number there, or if it's virtual, skip it.
6048                 osrfHash* _f = osrfHashGet( fields, (char*) columnName );
6049                 if( _f ) {
6050
6051                         if ( str_is_true( osrfHashGet(_f, "virtual") ) )
6052                                 continue;   // skip this column: IDL says it's virtual
6053
6054                         const char* pos = (char*)osrfHashGet(_f, "array_position");
6055                         if ( !pos )      // IDL has no sequence number for it.  This shouldn't happen,
6056                                 continue;    // since we assign sequence numbers dynamically as we load the IDL.
6057
6058                         fmIndex = atoi( pos );
6059                         osrfLogInternal(OSRF_LOG_MARK, "... Found column at position [%s]...", pos);
6060                 } else {
6061                         continue;     // This field is not defined in the IDL
6062                 }
6063
6064                 // Stuff the column value into a slot in the JSON_ARRAY, indexed according to the
6065                 // sequence number from the IDL (which is likely to be different from the sequence
6066                 // of columns in the SELECT clause).
6067                 if (dbi_result_field_is_null_idx(result, columnIndex)) {
6068                         jsonObjectSetIndex( object, fmIndex, jsonNewObject(NULL) );
6069                 } else {
6070
6071                         switch( type ) {
6072
6073                                 case DBI_TYPE_INTEGER :
6074
6075                                         if( attr & DBI_INTEGER_SIZE8 )
6076                                                 jsonObjectSetIndex( object, fmIndex,
6077                                                         jsonNewNumberObject(dbi_result_get_longlong_idx(result, columnIndex)));
6078                                         else
6079                                                 jsonObjectSetIndex( object, fmIndex,
6080                                                         jsonNewNumberObject(dbi_result_get_int_idx(result, columnIndex)));
6081
6082                                         break;
6083
6084                                 case DBI_TYPE_DECIMAL :
6085                                         jsonObjectSetIndex( object, fmIndex,
6086                                                         jsonNewNumberObject(dbi_result_get_double_idx(result, columnIndex)));
6087                                         break;
6088
6089                                 case DBI_TYPE_STRING :
6090
6091                                         jsonObjectSetIndex(
6092                                                 object,
6093                                                 fmIndex,
6094                                                 jsonNewObject( dbi_result_get_string_idx(result, columnIndex) )
6095                                         );
6096
6097                                         break;
6098
6099                                 case DBI_TYPE_DATETIME : {
6100
6101                                         char dt_string[256] = "";
6102                                         struct tm gmdt;
6103
6104                                         // Fetch the date column as a time_t
6105                                         time_t _tmp_dt = dbi_result_get_datetime_idx(result, columnIndex);
6106
6107                                         // Translate the time_t to a human-readable string
6108                                         if (!(attr & DBI_DATETIME_DATE)) {
6109                                                 gmtime_r( &_tmp_dt, &gmdt );
6110                                                 strftime(dt_string, sizeof(dt_string), "%T", &gmdt);
6111                                         } else if (!(attr & DBI_DATETIME_TIME)) {
6112                                                 localtime_r( &_tmp_dt, &gmdt );
6113                                                 strftime(dt_string, sizeof(dt_string), "%F", &gmdt);
6114                                         } else {
6115                                                 localtime_r( &_tmp_dt, &gmdt );
6116                                                 strftime(dt_string, sizeof(dt_string), "%FT%T%z", &gmdt);
6117                                         }
6118
6119                                         jsonObjectSetIndex( object, fmIndex, jsonNewObject(dt_string) );
6120
6121                                         break;
6122                                 }
6123                                 case DBI_TYPE_BINARY :
6124                                         osrfLogError( OSRF_LOG_MARK,
6125                                                 "Can't do binary at column %s : index %d", columnName, columnIndex);
6126                         } // End switch
6127                 }
6128                 ++columnIndex;
6129         } // End while
6130
6131         return object;
6132 }
6133
6134 static jsonObject* oilsMakeJSONFromResult( dbi_result result ) {
6135         if(!result) return jsonNULL;
6136
6137         jsonObject* object = jsonNewObject(NULL);
6138
6139         time_t _tmp_dt;
6140         char dt_string[256];
6141         struct tm gmdt;
6142
6143         int fmIndex;
6144         int columnIndex = 1;
6145         int attr;
6146         unsigned short type;
6147         const char* columnName;
6148
6149         /* cycle through the column list */
6150         while( (columnName = dbi_result_get_field_name(result, columnIndex)) ) {
6151
6152                 osrfLogInternal(OSRF_LOG_MARK, "Looking for column named [%s]...", (char*)columnName);
6153
6154                 fmIndex = -1; // reset the position
6155
6156                 /* determine the field type and storage attributes */
6157                 type = dbi_result_get_field_type_idx(result, columnIndex);
6158                 attr = dbi_result_get_field_attribs_idx(result, columnIndex);
6159
6160                 if (dbi_result_field_is_null_idx(result, columnIndex)) {
6161                         jsonObjectSetKey( object, columnName, jsonNewObject(NULL) );
6162                 } else {
6163
6164                         switch( type ) {
6165
6166                                 case DBI_TYPE_INTEGER :
6167
6168                                         if( attr & DBI_INTEGER_SIZE8 )
6169                                                 jsonObjectSetKey( object, columnName,
6170                                                                 jsonNewNumberObject(dbi_result_get_longlong_idx(
6171                                                                                 result, columnIndex)) );
6172                                         else
6173                                                 jsonObjectSetKey( object, columnName,
6174                                                                 jsonNewNumberObject(dbi_result_get_int_idx(result, columnIndex)) );
6175                                         break;
6176
6177                                 case DBI_TYPE_DECIMAL :
6178                                         jsonObjectSetKey( object, columnName,
6179                                                         jsonNewNumberObject(dbi_result_get_double_idx(result, columnIndex)) );
6180                                         break;
6181
6182                                 case DBI_TYPE_STRING :
6183                                         jsonObjectSetKey( object, columnName,
6184                                                         jsonNewObject(dbi_result_get_string_idx(result, columnIndex)) );
6185                                         break;
6186
6187                                 case DBI_TYPE_DATETIME :
6188
6189                                         memset(dt_string, '\0', sizeof(dt_string));
6190                                         memset(&gmdt, '\0', sizeof(gmdt));
6191
6192                                         _tmp_dt = dbi_result_get_datetime_idx(result, columnIndex);
6193
6194
6195                                         if (!(attr & DBI_DATETIME_DATE)) {
6196                                                 gmtime_r( &_tmp_dt, &gmdt );
6197                                                 strftime(dt_string, sizeof(dt_string), "%T", &gmdt);
6198                                         } else if (!(attr & DBI_DATETIME_TIME)) {
6199                                                 localtime_r( &_tmp_dt, &gmdt );
6200                                                 strftime(dt_string, sizeof(dt_string), "%F", &gmdt);
6201                                         } else {
6202                                                 localtime_r( &_tmp_dt, &gmdt );
6203                                                 strftime(dt_string, sizeof(dt_string), "%FT%T%z", &gmdt);
6204                                         }
6205
6206                                         jsonObjectSetKey( object, columnName, jsonNewObject(dt_string) );
6207                                         break;
6208
6209                                 case DBI_TYPE_BINARY :
6210                                         osrfLogError( OSRF_LOG_MARK,
6211                                                 "Can't do binary at column %s : index %d", columnName, columnIndex );
6212                         }
6213                 }
6214                 ++columnIndex;
6215         } // end while loop traversing result
6216
6217         return object;
6218 }
6219
6220 // Interpret a string as true or false
6221 static int str_is_true( const char* str ) {
6222         if( NULL == str || strcasecmp( str, "true" ) )
6223                 return 0;
6224         else
6225                 return 1;
6226 }
6227
6228 // Interpret a jsonObject as true or false
6229 static int obj_is_true( const jsonObject* obj ) {
6230         if( !obj )
6231                 return 0;
6232         else switch( obj->type )
6233         {
6234                 case JSON_BOOL :
6235                         if( obj->value.b )
6236                                 return 1;
6237                         else
6238                                 return 0;
6239                 case JSON_STRING :
6240                         if( strcasecmp( obj->value.s, "true" ) )
6241                                 return 0;
6242                         else
6243                                 return 1;
6244                 case JSON_NUMBER :          // Support 1/0 for perl's sake
6245                         if( jsonObjectGetNumber( obj ) == 1.0 )
6246                                 return 1;
6247                         else
6248                                 return 0;
6249                 default :
6250                         return 0;
6251         }
6252 }
6253
6254 // Translate a numeric code into a text string identifying a type of
6255 // jsonObject.  To be used for building error messages.
6256 static const char* json_type( int code ) {
6257         switch ( code )
6258         {
6259                 case 0 :
6260                         return "JSON_HASH";
6261                 case 1 :
6262                         return "JSON_ARRAY";
6263                 case 2 :
6264                         return "JSON_STRING";
6265                 case 3 :
6266                         return "JSON_NUMBER";
6267                 case 4 :
6268                         return "JSON_NULL";
6269                 case 5 :
6270                         return "JSON_BOOL";
6271                 default :
6272                         return "(unrecognized)";
6273         }
6274 }
6275
6276 // Extract the "primitive" attribute from an IDL field definition.
6277 // If we haven't initialized the app, then we must be running in
6278 // some kind of testbed.  In that case, default to "string".
6279 static const char* get_primitive( osrfHash* field ) {
6280         const char* s = osrfHashGet( field, "primitive" );
6281         if( !s ) {
6282                 if( child_initialized )
6283                         osrfLogError(
6284                                 OSRF_LOG_MARK,
6285                                 "%s ERROR No \"datatype\" attribute for field \"%s\"",
6286                                 modulename,
6287                                 osrfHashGet( field, "name" )
6288                         );
6289
6290                 s = "string";
6291         }
6292         return s;
6293 }
6294
6295 // Extract the "datatype" attribute from an IDL field definition.
6296 // If we haven't initialized the app, then we must be running in
6297 // some kind of testbed.  In that case, default to to NUMERIC,
6298 // since we look at the datatype only for numbers.
6299 static const char* get_datatype( osrfHash* field ) {
6300         const char* s = osrfHashGet( field, "datatype" );
6301         if( !s ) {
6302                 if( child_initialized )
6303                         osrfLogError(
6304                                 OSRF_LOG_MARK,
6305                                 "%s ERROR No \"datatype\" attribute for field \"%s\"",
6306                                 modulename,
6307                                 osrfHashGet( field, "name" )
6308                         );
6309                 else
6310                         s = "NUMERIC";
6311         }
6312         return s;
6313 }
6314
6315 /**
6316         @brief Determine whether a string is potentially a valid SQL identifier.
6317         @param s The identifier to be tested.
6318         @return 1 if the input string is potentially a valid SQL identifier, or 0 if not.
6319
6320         Purpose: to prevent certain kinds of SQL injection.  To that end we don't necessarily
6321         need to follow all the rules exactly, such as requiring that the first character not
6322         be a digit.
6323
6324         We allow leading and trailing white space.  In between, we do not allow punctuation
6325         (except for underscores and dollar signs), control characters, or embedded white space.
6326
6327         More pedantically we should allow quoted identifiers containing arbitrary characters, but
6328         for the foreseeable future such quoted identifiers are not likely to be an issue.
6329 */
6330 static int is_identifier( const char* s) {
6331         if( !s )
6332                 return 0;
6333
6334         // Skip leading white space
6335         while( isspace( (unsigned char) *s ) )
6336                 ++s;
6337
6338         if( !s )
6339                 return 0;   // Nothing but white space?  Not okay.
6340
6341         // Check each character until we reach white space or
6342         // end-of-string.  Letters, digits, underscores, and
6343         // dollar signs are okay. With the exception of periods
6344         // (as in schema.identifier), control characters and other
6345         // punctuation characters are not okay.  Anything else
6346         // is okay -- it could for example be part of a multibyte
6347         // UTF8 character such as a letter with diacritical marks,
6348         // and those are allowed.
6349         do {
6350                 if( isalnum( (unsigned char) *s )
6351                         || '.' == *s
6352                         || '_' == *s
6353                         || '$' == *s )
6354                         ;  // Fine; keep going
6355                 else if(   ispunct( (unsigned char) *s )
6356                                 || iscntrl( (unsigned char) *s ) )
6357                         return 0;
6358                         ++s;
6359         } while( *s && ! isspace( (unsigned char) *s ) );
6360
6361         // If we found any white space in the above loop,
6362         // the rest had better be all white space.
6363
6364         while( isspace( (unsigned char) *s ) )
6365                 ++s;
6366
6367         if( *s )
6368                 return 0;   // White space was embedded within non-white space
6369
6370         return 1;
6371 }
6372
6373 /**
6374         @brief Determine whether to accept a character string as a comparison operator.
6375         @param op The candidate comparison operator.
6376         @return 1 if the string is acceptable as a comparison operator, or 0 if not.
6377
6378         We don't validate the operator for real.  We just make sure that it doesn't contain
6379         any semicolons or white space (with special exceptions for a few specific operators).
6380         The idea is to block certain kinds of SQL injection.  If it has no semicolons or white
6381         space but it's still not a valid operator, then the database will complain.
6382
6383         Another approach would be to compare the string against a short list of approved operators.
6384         We don't do that because we want to allow custom operators like ">100*", which at this
6385         writing would be difficult or impossible to express otherwise in a JSON query.
6386 */
6387 static int is_good_operator( const char* op ) {
6388         if( !op ) return 0;   // Sanity check
6389
6390         const char* s = op;
6391         while( *s ) {
6392                 if( isspace( (unsigned char) *s ) ) {
6393                         // Special exceptions for SIMILAR TO, IS DISTINCT FROM,
6394                         // and IS NOT DISTINCT FROM.
6395                         if( !strcasecmp( op, "similar to" ) )
6396                                 return 1;
6397                         else if( !strcasecmp( op, "is distinct from" ) )
6398                                 return 1;
6399                         else if( !strcasecmp( op, "is not distinct from" ) )
6400                                 return 1;
6401                         else
6402                                 return 0;
6403                 }
6404                 else if( ';' == *s )
6405                         return 0;
6406                 ++s;
6407         }
6408         return 1;
6409 }
6410
6411 /**
6412         @name Query Frame Management
6413
6414         The following machinery supports a stack of query frames for use by SELECT().
6415
6416         A query frame caches information about one level of a SELECT query.  When we enter
6417         a subquery, we push another query frame onto the stack, and pop it off when we leave.
6418
6419         The query frame stores information about the core class, and about any joined classes
6420         in the FROM clause.
6421
6422         The main purpose is to map table aliases to classes and tables, so that a query can
6423         join to the same table more than once.  A secondary goal is to reduce the number of
6424         lookups in the IDL by caching the results.
6425 */
6426 /*@{*/
6427
6428 #define STATIC_CLASS_INFO_COUNT 3
6429
6430 static ClassInfo static_class_info[ STATIC_CLASS_INFO_COUNT ];
6431
6432 /**
6433         @brief Allocate a ClassInfo as raw memory.
6434         @return Pointer to the newly allocated ClassInfo.
6435
6436         Except for the in_use flag, which is used only by the allocation and deallocation
6437         logic, we don't initialize the ClassInfo here.
6438 */
6439 static ClassInfo* allocate_class_info( void ) {
6440         // In order to reduce the number of mallocs and frees, we return a static
6441         // instance of ClassInfo, if we can find one that we're not already using.
6442         // We rely on the fact that the compiler will implicitly initialize the
6443         // static instances so that in_use == 0.
6444
6445         int i;
6446         for( i = 0; i < STATIC_CLASS_INFO_COUNT; ++i ) {
6447                 if( ! static_class_info[ i ].in_use ) {
6448                         static_class_info[ i ].in_use = 1;
6449                         return static_class_info + i;
6450                 }
6451         }
6452
6453         // The static ones are all in use.  Malloc one.
6454
6455         return safe_malloc( sizeof( ClassInfo ) );
6456 }
6457
6458 /**
6459         @brief Free any malloc'd memory owned by a ClassInfo, returning it to a pristine state.
6460         @param info Pointer to the ClassInfo to be cleared.
6461 */
6462 static void clear_class_info( ClassInfo* info ) {
6463         // Sanity check
6464         if( ! info )
6465                 return;
6466
6467         // Free any malloc'd strings
6468
6469         if( info->alias != info->alias_store )
6470                 free( info->alias );
6471
6472         if( info->class_name != info->class_name_store )
6473                 free( info->class_name );
6474
6475         free( info->source_def );
6476
6477         info->alias = info->class_name = info->source_def = NULL;
6478         info->next = NULL;
6479 }
6480
6481 /**
6482         @brief Free a ClassInfo and everything it owns.
6483         @param info Pointer to the ClassInfo to be freed.
6484 */
6485 static void free_class_info( ClassInfo* info ) {
6486         // Sanity check
6487         if( ! info )
6488                 return;
6489
6490         clear_class_info( info );
6491
6492         // If it's one of the static instances, just mark it as not in use
6493
6494         int i;
6495         for( i = 0; i < STATIC_CLASS_INFO_COUNT; ++i ) {
6496                 if( info == static_class_info + i ) {
6497                         static_class_info[ i ].in_use = 0;
6498                         return;
6499                 }
6500         }
6501
6502         // Otherwise it must have been malloc'd, so free it
6503
6504         free( info );
6505 }
6506
6507 /**
6508         @brief Populate an already-allocated ClassInfo.
6509         @param info Pointer to the ClassInfo to be populated.
6510         @param alias Alias for the class.  If it is NULL, or an empty string, use the class
6511         name for an alias.
6512         @param class Name of the class.
6513         @return Zero if successful, or 1 if not.
6514
6515         Populate the ClassInfo with copies of the alias and class name, and with pointers to
6516         the relevant portions of the IDL for the specified class.
6517 */
6518 static int build_class_info( ClassInfo* info, const char* alias, const char* class ) {
6519         // Sanity checks
6520         if( ! info ){
6521                 osrfLogError( OSRF_LOG_MARK,
6522                                           "%s ERROR: No ClassInfo available to populate", modulename );
6523                 info->alias = info->class_name = info->source_def = NULL;
6524                 info->class_def = info->fields = info->links = NULL;
6525                 return 1;
6526         }
6527
6528         if( ! class ) {
6529                 osrfLogError( OSRF_LOG_MARK,
6530                                           "%s ERROR: No class name provided for lookup", modulename );
6531                 info->alias = info->class_name = info->source_def = NULL;
6532                 info->class_def = info->fields = info->links = NULL;
6533                 return 1;
6534         }
6535
6536         // Alias defaults to class name if not supplied
6537         if( ! alias || ! alias[ 0 ] )
6538                 alias = class;
6539
6540         // Look up class info in the IDL
6541         osrfHash* class_def = osrfHashGet( oilsIDL(), class );
6542         if( ! class_def ) {
6543                 osrfLogError( OSRF_LOG_MARK,
6544                                           "%s ERROR: Class %s not defined in IDL", modulename, class );
6545                 info->alias = info->class_name = info->source_def = NULL;
6546                 info->class_def = info->fields = info->links = NULL;
6547                 return 1;
6548         } else if( str_is_true( osrfHashGet( class_def, "virtual" ) ) ) {
6549                 osrfLogError( OSRF_LOG_MARK,
6550                                           "%s ERROR: Class %s is defined as virtual", modulename, class );
6551                 info->alias = info->class_name = info->source_def = NULL;
6552                 info->class_def = info->fields = info->links = NULL;
6553                 return 1;
6554         }
6555
6556         osrfHash* links = osrfHashGet( class_def, "links" );
6557         if( ! links ) {
6558                 osrfLogError( OSRF_LOG_MARK,
6559                                           "%s ERROR: No links defined in IDL for class %s", modulename, class );
6560                 info->alias = info->class_name = info->source_def = NULL;
6561                 info->class_def = info->fields = info->links = NULL;
6562                 return 1;
6563         }
6564
6565         osrfHash* fields = osrfHashGet( class_def, "fields" );
6566         if( ! fields ) {
6567                 osrfLogError( OSRF_LOG_MARK,
6568                                           "%s ERROR: No fields defined in IDL for class %s", modulename, class );
6569                 info->alias = info->class_name = info->source_def = NULL;
6570                 info->class_def = info->fields = info->links = NULL;
6571                 return 1;
6572         }
6573
6574         char* source_def = getRelation( class_def );
6575         if( ! source_def )
6576                 return 1;
6577
6578         // We got everything we need, so populate the ClassInfo
6579         if( strlen( alias ) > ALIAS_STORE_SIZE )
6580                 info->alias = strdup( alias );
6581         else {
6582                 strcpy( info->alias_store, alias );
6583                 info->alias = info->alias_store;
6584         }
6585
6586         if( strlen( class ) > CLASS_NAME_STORE_SIZE )
6587                 info->class_name = strdup( class );
6588         else {
6589                 strcpy( info->class_name_store, class );
6590                 info->class_name = info->class_name_store;
6591         }
6592
6593         info->source_def = source_def;
6594
6595         info->class_def = class_def;
6596         info->links     = links;
6597         info->fields    = fields;
6598
6599         return 0;
6600 }
6601
6602 #define STATIC_FRAME_COUNT 3
6603
6604 static QueryFrame static_frame[ STATIC_FRAME_COUNT ];
6605
6606 /**
6607         @brief Allocate a QueryFrame as raw memory.
6608         @return Pointer to the newly allocated QueryFrame.
6609
6610         Except for the in_use flag, which is used only by the allocation and deallocation
6611         logic, we don't initialize the QueryFrame here.
6612 */
6613 static QueryFrame* allocate_frame( void ) {
6614         // In order to reduce the number of mallocs and frees, we return a static
6615         // instance of QueryFrame, if we can find one that we're not already using.
6616         // We rely on the fact that the compiler will implicitly initialize the
6617         // static instances so that in_use == 0.
6618
6619         int i;
6620         for( i = 0; i < STATIC_FRAME_COUNT; ++i ) {
6621                 if( ! static_frame[ i ].in_use ) {
6622                         static_frame[ i ].in_use = 1;
6623                         return static_frame + i;
6624                 }
6625         }
6626
6627         // The static ones are all in use.  Malloc one.
6628
6629         return safe_malloc( sizeof( QueryFrame ) );
6630 }
6631
6632 /**
6633         @brief Free a QueryFrame, and all the memory it owns.
6634         @param frame Pointer to the QueryFrame to be freed.
6635 */
6636 static void free_query_frame( QueryFrame* frame ) {
6637         // Sanity check
6638         if( ! frame )
6639                 return;
6640
6641         clear_class_info( &frame->core );
6642
6643         // Free the join list
6644         ClassInfo* temp;
6645         ClassInfo* info = frame->join_list;
6646         while( info ) {
6647                 temp = info->next;
6648                 free_class_info( info );
6649                 info = temp;
6650         }
6651
6652         frame->join_list = NULL;
6653         frame->next = NULL;
6654
6655         // If the frame is a static instance, just mark it as unused
6656         int i;
6657         for( i = 0; i < STATIC_FRAME_COUNT; ++i ) {
6658                 if( frame == static_frame + i ) {
6659                         static_frame[ i ].in_use = 0;
6660                         return;
6661                 }
6662         }
6663
6664         // Otherwise it must have been malloc'd, so free it
6665
6666         free( frame );
6667 }
6668
6669 /**
6670         @brief Search a given QueryFrame for a specified alias.
6671         @param frame Pointer to the QueryFrame to be searched.
6672         @param target The alias for which to search.
6673         @return Pointer to the ClassInfo for the specified alias, if found; otherwise NULL.
6674 */
6675 static ClassInfo* search_alias_in_frame( QueryFrame* frame, const char* target ) {
6676         if( ! frame || ! target ) {
6677                 return NULL;
6678         }
6679
6680         ClassInfo* found_class = NULL;
6681
6682         if( !strcmp( target, frame->core.alias ) )
6683                 return &(frame->core);
6684         else {
6685                 ClassInfo* curr_class = frame->join_list;
6686                 while( curr_class ) {
6687                         if( strcmp( target, curr_class->alias ) )
6688                                 curr_class = curr_class->next;
6689                         else {
6690                                 found_class = curr_class;
6691                                 break;
6692                         }
6693                 }
6694         }
6695
6696         return found_class;
6697 }
6698
6699 /**
6700         @brief Push a new (blank) QueryFrame onto the stack.
6701 */
6702 static void push_query_frame( void ) {
6703         QueryFrame* frame = allocate_frame();
6704         frame->join_list = NULL;
6705         frame->next = curr_query;
6706
6707         // Initialize the ClassInfo for the core class
6708         ClassInfo* core = &frame->core;
6709         core->alias = core->class_name = core->source_def = NULL;
6710         core->class_def = core->fields = core->links = NULL;
6711
6712         curr_query = frame;
6713 }
6714
6715 /**
6716         @brief Pop a QueryFrame off the stack and destroy it.
6717 */
6718 static void pop_query_frame( void ) {
6719         // Sanity check
6720         if( ! curr_query )
6721                 return;
6722
6723         QueryFrame* popped = curr_query;
6724         curr_query = popped->next;
6725
6726         free_query_frame( popped );
6727 }
6728
6729 /**
6730         @brief Populate the ClassInfo for the core class.
6731         @param alias Alias for the core class.  If it is NULL or an empty string, we use the
6732         class name as an alias.
6733         @param class_name Name of the core class.
6734         @return Zero if successful, or 1 if not.
6735
6736         Populate the ClassInfo of the core class with copies of the alias and class name, and
6737         with pointers to the relevant portions of the IDL for the core class.
6738 */
6739 static int add_query_core( const char* alias, const char* class_name ) {
6740
6741         // Sanity checks
6742         if( ! curr_query ) {
6743                 osrfLogError( OSRF_LOG_MARK,
6744                                           "%s ERROR: No QueryFrame available for class %s", modulename, class_name );
6745                 return 1;
6746         } else if( curr_query->core.alias ) {
6747                 osrfLogError( OSRF_LOG_MARK,
6748                                           "%s ERROR: Core class %s already populated as %s",
6749                                           modulename, curr_query->core.class_name, curr_query->core.alias );
6750                 return 1;
6751         }
6752
6753         build_class_info( &curr_query->core, alias, class_name );
6754         if( curr_query->core.alias )
6755                 return 0;
6756         else {
6757                 osrfLogError( OSRF_LOG_MARK,
6758                                           "%s ERROR: Unable to look up core class %s", modulename, class_name );
6759                 return 1;
6760         }
6761 }
6762
6763 /**
6764         @brief Search the current QueryFrame for a specified alias.
6765         @param target The alias for which to search.
6766         @return A pointer to the corresponding ClassInfo, if found; otherwise NULL.
6767 */
6768 static inline ClassInfo* search_alias( const char* target ) {
6769         return search_alias_in_frame( curr_query, target );
6770 }
6771
6772 /**
6773         @brief Search all levels of query for a specified alias, starting with the current query.
6774         @param target The alias for which to search.
6775         @return A pointer to the corresponding ClassInfo, if found; otherwise NULL.
6776 */
6777 static ClassInfo* search_all_alias( const char* target ) {
6778         ClassInfo* found_class = NULL;
6779         QueryFrame* curr_frame = curr_query;
6780
6781         while( curr_frame ) {
6782                 if(( found_class = search_alias_in_frame( curr_frame, target ) ))
6783                         break;
6784                 else
6785                         curr_frame = curr_frame->next;
6786         }
6787
6788         return found_class;
6789 }
6790
6791 /**
6792         @brief Add a class to the list of classes joined to the current query.
6793         @param alias Alias of the class to be added.  If it is NULL or an empty string, we use
6794         the class name as an alias.
6795         @param classname The name of the class to be added.
6796         @return A pointer to the ClassInfo for the added class, if successful; otherwise NULL.
6797 */
6798 static ClassInfo* add_joined_class( const char* alias, const char* classname ) {
6799
6800         if( ! classname || ! *classname ) {    // sanity check
6801                 osrfLogError( OSRF_LOG_MARK, "Can't join a class with no class name" );
6802                 return NULL;
6803         }
6804
6805         if( ! alias )
6806                 alias = classname;
6807
6808         const ClassInfo* conflict = search_alias( alias );
6809         if( conflict ) {
6810                 osrfLogError( OSRF_LOG_MARK,
6811                                           "%s ERROR: Table alias \"%s\" conflicts with class \"%s\"",
6812                                           modulename, alias, conflict->class_name );
6813                 return NULL;
6814         }
6815
6816         ClassInfo* info = allocate_class_info();
6817
6818         if( build_class_info( info, alias, classname ) ) {
6819                 free_class_info( info );
6820                 return NULL;
6821         }
6822
6823         // Add the new ClassInfo to the join list of the current QueryFrame
6824         info->next = curr_query->join_list;
6825         curr_query->join_list = info;
6826
6827         return info;
6828 }
6829
6830 /**
6831         @brief Destroy all nodes on the query stack.
6832 */
6833 static void clear_query_stack( void ) {
6834         while( curr_query )
6835                 pop_query_frame();
6836 }
6837
6838 /*@}*/