src/modules/extra/m_pgsql.cpp

   1 /*       +------------------------------------+
   2  *       | Inspire Internet Relay Chat Daemon |
   3  *       +------------------------------------+
   4  *
   5  *  InspIRCd is copyright (C) 2002-2004 ChatSpike-Dev.
   6  *                       E-mail:
   7  *                <brain@chatspike.net>
   8  *                <Craig@chatspike.net>
   9  *                <omster@gmail.com>
  10  *
  11  * Written by Craig Edwards, Craig McLure, and others.
  12  * This program is free but copyrighted software; see
  13  *            the file COPYING for details.
  14  *
  15  * ---------------------------------------------------
  16  */
  17
  18 #include <cstdlib>
  19 #include <sstream>
  20 #include <string>
  21 #include <deque>
  22 #include <map>
  23 #include <libpq-fe.h>
  24
  25 #include "users.h"
  26 #include "channels.h"
  27 #include "modules.h"
  28 #include "helperfuncs.h"
  29 #include "inspircd.h"
  30 #include "configreader.h"
  31
  32 #include "m_sqlv2.h"
  33
  34 /* $ModDesc: PostgreSQL Service Provider module for all other m_sql* modules, uses v2 of the SQL API */
  35 /* $CompileFlags: -I`pg_config --includedir` `perl extra/pgsql_config.pl` */
  36 /* $LinkerFlags: -L`pg_config --libdir` -lpq */
  37
  38 /* UGH, UGH, UGH, UGH, UGH, UGH
  39  * I'm having trouble seeing how I
  40  * can avoid this. The core-defined
  41  * constructors for InspSocket just
  42  * aren't suitable...and if I'm
  43  * reimplementing them I need this so
  44  * I can access the socket engine :\
  45  */
  46 extern InspIRCd* ServerInstance;
  47 InspSocket* socket_ref[MAX_DESCRIPTORS];
  48
  49 /* Forward declare, so we can have the typedef neatly at the top */
  50 class SQLConn;
  51 /* Also needs forward declaration, as it's used inside SQLconn */
  52 class ModulePgSQL;
  53
  54 typedef std::map<std::string, SQLConn*> ConnMap;
  55
  56 /* CREAD,       Connecting and wants read event
  57  * CWRITE,      Connecting and wants write event
  58  * WREAD,       Connected/Working and wants read event
  59  * WWRITE,      Connected/Working and wants write event
  60  */
  61 enum SQLstatus { CREAD, CWRITE, WREAD, WWRITE };
  62
  63 /** QueryQueue, a queue of queries waiting to be executed.
  64  * This maintains two queues internally, one for 'priority'
  65  * queries and one for less important ones. Each queue has
  66  * new queries appended to it and ones to execute are popped
  67  * off the front. This keeps them flowing round nicely and no
  68  * query should ever get 'stuck' for too long. If there are
  69  * queries in the priority queue they will be executed first,
  70  * 'unimportant' queries will only be executed when the
  71  * priority queue is empty.
  72  *
  73  * We store lists of SQLrequest's here, by value as we want to avoid storing
  74  * any data allocated inside the client module (in case that module is unloaded
  75  * while the query is in progress).
  76  *
  77  * Because we want to work on the current SQLrequest in-situ, we need a way
  78  * of accessing the request we are currently processing, QueryQueue::front(),
  79  * but that call needs to always return the same request until that request
  80  * is removed from the queue, this is what the 'which' variable is. New queries are
  81  * always added to the back of one of the two queues, but if when front()
  82  * is first called then the priority queue is empty then front() will return
  83  * a query from the normal queue, but if a query is then added to the priority
  84  * queue then front() must continue to return the front of the *normal* queue
  85  * until pop() is called.
  86  */
  87
  88 class QueryQueue : public classbase
  89 {
  90 private:
  91         typedef std::deque<SQLrequest> ReqDeque;
  92
  93         ReqDeque priority;      /* The priority queue */
  94         ReqDeque normal;        /* The 'normal' queue */
  95         enum { PRI, NOR, NON } which;   /* Which queue the currently active element is at the front of */
  96
  97 public:
  98         QueryQueue()
  99         : which(NON)
 100         {
 101         }
 102
 103         void push(const SQLrequest &q)
 104         {
 105                 log(DEBUG, "QueryQueue::push(): Adding %s query to queue: %s", ((q.pri) ? "priority" : "non-priority"), q.query.q.c_str());
 106
 107                 if(q.pri)
 108                         priority.push_back(q);
 109                 else
 110                         normal.push_back(q);
 111         }
 112
 113         void pop()
 114         {
 115                 if((which == PRI) && priority.size())
 116                 {
 117                         priority.pop_front();
 118                 }
 119                 else if((which == NOR) && normal.size())
 120                 {
 121                         normal.pop_front();
 122                 }
 123
 124                 /* Reset this */
 125                 which = NON;
 126
 127                 /* Silently do nothing if there was no element to pop() */
 128         }
 129
 130         SQLrequest& front()
 131         {
 132                 switch(which)
 133                 {
 134                         case PRI:
 135                                 return priority.front();
 136                         case NOR:
 137                                 return normal.front();
 138                         default:
 139                                 if(priority.size())
 140                                 {
 141                                         which = PRI;
 142                                         return priority.front();
 143                                 }
 144
 145                                 if(normal.size())
 146                                 {
 147                                         which = NOR;
 148                                         return normal.front();
 149                                 }
 150
 151                                 /* This will probably result in a segfault,
 152                                  * but the caller should have checked totalsize()
 153                                  * first so..meh - moron :p
 154                                  */
 155
 156                                 return priority.front();
 157                 }
 158         }
 159
 160         std::pair<int, int> size()
 161         {
 162                 return std::make_pair(priority.size(), normal.size());
 163         }
 164
 165         int totalsize()
 166         {
 167                 return priority.size() + normal.size();
 168         }
 169
 170         void PurgeModule(Module* mod)
 171         {
 172                 DoPurgeModule(mod, priority);
 173                 DoPurgeModule(mod, normal);
 174         }
 175
 176 private:
 177         void DoPurgeModule(Module* mod, ReqDeque& q)
 178         {
 179                 for(ReqDeque::iterator iter = q.begin(); iter != q.end(); iter++)
 180                 {
 181                         if(iter->GetSource() == mod)
 182                         {
 183                                 if(iter->id == front().id)
 184                                 {
 185                                         /* It's the currently active query.. :x */
 186                                         iter->SetSource(NULL);
 187                                 }
 188                                 else
 189                                 {
 190                                         /* It hasn't been executed yet..just remove it */
 191                                         iter = q.erase(iter);
 192                                 }
 193                         }
 194                 }
 195         }
 196 };
 197
 198 /** PgSQLresult is a subclass of the mostly-pure-virtual class SQLresult.
 199  * All SQL providers must create their own subclass and define it's methods using that
 200  * database library's data retriveal functions. The aim is to avoid a slow and inefficient process
 201  * of converting all data to a common format before it reaches the result structure. This way
 202  * data is passes to the module nearly as directly as if it was using the API directly itself.
 203  */
 204
 205 class PgSQLresult : public SQLresult
 206 {
 207         PGresult* res;
 208         int currentrow;
 209         int rows;
 210         int cols;
 211
 212         SQLfieldList* fieldlist;
 213         SQLfieldMap* fieldmap;
 214 public:
 215         PgSQLresult(Module* self, Module* to, unsigned long id, PGresult* result)
 216         : SQLresult(self, to, id), res(result), currentrow(0), fieldlist(NULL), fieldmap(NULL)
 217         {
 218                 rows = PQntuples(res);
 219                 cols = PQnfields(res);
 220
 221                 log(DEBUG, "Created new PgSQL result; %d rows, %d columns, %s affected", rows, cols, PQcmdTuples(res));
 222         }
 223
 224         ~PgSQLresult()
 225         {
 226                 /* If we allocated these, free them... */
 227                 if(fieldlist)
 228                         DELETE(fieldlist);
 229
 230                 if(fieldmap)
 231                         DELETE(fieldmap);
 232
 233                 PQclear(res);
 234         }
 235
 236         virtual int Rows()
 237         {
 238                 if(!cols && !rows)
 239                 {
 240                         return atoi(PQcmdTuples(res));
 241                 }
 242                 else
 243                 {
 244                         return rows;
 245                 }
 246         }
 247
 248         virtual int Cols()
 249         {
 250                 return PQnfields(res);
 251         }
 252
 253         virtual std::string ColName(int column)
 254         {
 255                 char* name = PQfname(res, column);
 256
 257                 return (name) ? name : "";
 258         }
 259
 260         virtual int ColNum(const std::string &column)
 261         {
 262                 int n = PQfnumber(res, column.c_str());
 263
 264                 if(n == -1)
 265                 {
 266                         throw SQLbadColName();
 267                 }
 268                 else
 269                 {
 270                         return n;
 271                 }
 272         }
 273
 274         virtual SQLfield GetValue(int row, int column)
 275         {
 276                 char* v = PQgetvalue(res, row, column);
 277
 278                 if(v)
 279                 {
 280                         return SQLfield(std::string(v, PQgetlength(res, row, column)), PQgetisnull(res, row, column));
 281                 }
 282                 else
 283                 {
 284                         log(DEBUG, "PQgetvalue returned a null pointer..nobody wants to tell us what this means");
 285                         throw SQLbadColName();
 286                 }
 287         }
 288
 289         virtual SQLfieldList& GetRow()
 290         {
 291                 /* In an effort to reduce overhead we don't actually allocate the list
 292                  * until the first time it's needed...so...
 293                  */
 294                 if(fieldlist)
 295                 {
 296                         fieldlist->clear();
 297                 }
 298                 else
 299                 {
 300                         fieldlist = new SQLfieldList;
 301                 }
 302
 303                 if(currentrow < PQntuples(res))
 304                 {
 305                         int cols = PQnfields(res);
 306
 307                         for(int i = 0; i < cols; i++)
 308                         {
 309                                 fieldlist->push_back(GetValue(currentrow, i));
 310                         }
 311
 312                         currentrow++;
 313                 }
 314
 315                 return *fieldlist;
 316         }
 317
 318         virtual SQLfieldMap& GetRowMap()
 319         {
 320                 /* In an effort to reduce overhead we don't actually allocate the map
 321                  * until the first time it's needed...so...
 322                  */
 323                 if(fieldmap)
 324                 {
 325                         fieldmap->clear();
 326                 }
 327                 else
 328                 {
 329                         fieldmap = new SQLfieldMap;
 330                 }
 331
 332                 if(currentrow < PQntuples(res))
 333                 {
 334                         int cols = PQnfields(res);
 335
 336                         for(int i = 0; i < cols; i++)
 337                         {
 338                                 fieldmap->insert(std::make_pair(ColName(i), GetValue(currentrow, i)));
 339                         }
 340
 341                         currentrow++;
 342                 }
 343
 344                 return *fieldmap;
 345         }
 346
 347         virtual SQLfieldList* GetRowPtr()
 348         {
 349                 SQLfieldList* fl = new SQLfieldList;
 350
 351                 if(currentrow < PQntuples(res))
 352                 {
 353                         int cols = PQnfields(res);
 354
 355                         for(int i = 0; i < cols; i++)
 356                         {
 357                                 fl->push_back(GetValue(currentrow, i));
 358                         }
 359
 360                         currentrow++;
 361                 }
 362
 363                 return fl;
 364         }
 365
 366         virtual SQLfieldMap* GetRowMapPtr()
 367         {
 368                 SQLfieldMap* fm = new SQLfieldMap;
 369
 370                 if(currentrow < PQntuples(res))
 371                 {
 372                         int cols = PQnfields(res);
 373
 374                         for(int i = 0; i < cols; i++)
 375                         {
 376                                 fm->insert(std::make_pair(ColName(i), GetValue(currentrow, i)));
 377                         }
 378
 379                         currentrow++;
 380                 }
 381
 382                 return fm;
 383         }
 384
 385         virtual void Free(SQLfieldMap* fm)
 386         {
 387                 DELETE(fm);
 388         }
 389
 390         virtual void Free(SQLfieldList* fl)
 391         {
 392                 DELETE(fl);
 393         }
 394 };
 395
 396 /** SQLConn represents one SQL session.
 397  * Each session has its own persistent connection to the database.
 398  * This is a subclass of InspSocket so it can easily recieve read/write events from the core socket
 399  * engine, unlike the original MySQL module this module does not block. Ever. It gets a mild stabbing
 400  * if it dares to.
 401  */
 402
 403 class SQLConn : public InspSocket
 404 {
 405 private:
 406         ModulePgSQL* us;                /* Pointer to the SQL provider itself */
 407         Server* Srv;                    /* Server* for..uhm..something, maybe */
 408         std::string     dbhost; /* Database server hostname */
 409         unsigned int    dbport; /* Database server port */
 410         std::string     dbname; /* Database name */
 411         std::string     dbuser; /* Database username */
 412         std::string     dbpass; /* Database password */
 413         bool                    ssl;    /* If we should require SSL */
 414         PGconn*                 sql;    /* PgSQL database connection handle */
 415         SQLstatus               status; /* PgSQL database connection status */
 416         bool                    qinprog;/* If there is currently a query in progress */
 417         QueryQueue              queue;  /* Queue of queries waiting to be executed on this connection */
 418
 419 public:
 420
 421         /* This class should only ever be created inside this module, using this constructor, so we don't have to worry about the default ones */
 422
 423         SQLConn(ModulePgSQL* self, Server* srv, const std::string &h, unsigned int p, const std::string &d, const std::string &u, const std::string &pwd, bool s);
 424
 425         ~SQLConn();
 426
 427         bool DoResolve();
 428
 429         bool DoConnect();
 430
 431         virtual void Close();
 432
 433         bool DoPoll();
 434
 435         bool DoConnectedPoll();
 436
 437         void ShowStatus();
 438
 439         virtual bool OnDataReady();
 440
 441         virtual bool OnWriteReady();
 442
 443         virtual bool OnConnected();
 444
 445         bool DoEvent();
 446
 447         std::string MkInfoStr();
 448
 449         const char* StatusStr();
 450
 451         SQLerror DoQuery(SQLrequest &req);
 452
 453         SQLerror Query(const SQLrequest &req);
 454
 455         void OnUnloadModule(Module* mod);
 456 };
 457
 458 class ModulePgSQL : public Module
 459 {
 460 private:
 461         Server* Srv;
 462         ConnMap connections;
 463         unsigned long currid;
 464         char* sqlsuccess;
 465
 466 public:
 467         ModulePgSQL(Server* Me)
 468         : Module::Module(Me), Srv(Me), currid(0)
 469         {
 470                 log(DEBUG, "%s 'SQL' feature", Srv->PublishFeature("SQL", this) ? "Published" : "Couldn't publish");
 471
 472                 sqlsuccess = new char[strlen(SQLSUCCESS)+1];
 473
 474                 strcpy(sqlsuccess, SQLSUCCESS);
 475
 476                 OnRehash("");
 477         }
 478
 479         void Implements(char* List)
 480         {
 481                 List[I_OnUnloadModule] = List[I_OnRequest] = List[I_OnRehash] = List[I_OnUserRegister] = List[I_OnCheckReady] = List[I_OnUserDisconnect] = 1;
 482         }
 483
 484         virtual void OnRehash(const std::string &parameter)
 485         {
 486                 ConfigReader conf;
 487
 488                 /* Delete all the SQLConn objects in the connection lists,
 489                  * this will call their destructors where they can handle
 490                  * closing connections and such.
 491                  */
 492                 for(ConnMap::iterator iter = connections.begin(); iter != connections.end(); iter++)
 493                 {
 494                         DELETE(iter->second);
 495                 }
 496
 497                 /* Empty out our list of connections */
 498                 connections.clear();
 499
 500                 for(int i = 0; i < conf.Enumerate("database"); i++)
 501                 {
 502                         std::string id;
 503                         SQLConn* newconn;
 504
 505                         id = conf.ReadValue("database", "id", i);
 506                         newconn = new SQLConn(this, Srv,
 507                                                                                 conf.ReadValue("database", "hostname", i),
 508                                                                                 conf.ReadInteger("database", "port", i, true),
 509                                                                                 conf.ReadValue("database", "name", i),
 510                                                                                 conf.ReadValue("database", "username", i),
 511                                                                                 conf.ReadValue("database", "password", i),
 512                                                                                 conf.ReadFlag("database", "ssl", i));
 513
 514                         connections.insert(std::make_pair(id, newconn));
 515                 }
 516         }
 517
 518         virtual char* OnRequest(Request* request)
 519         {
 520                 if(strcmp(SQLREQID, request->GetId()) == 0)
 521                 {
 522                         SQLrequest* req = (SQLrequest*)request;
 523                         ConnMap::iterator iter;
 524
 525                         log(DEBUG, "Got query: '%s' with %d replacement parameters on id '%s'", req->query.q.c_str(), req->query.p.size(), req->dbid.c_str());
 526
 527                         if((iter = connections.find(req->dbid)) != connections.end())
 528                         {
 529                                 /* Execute query */
 530                                 req->id = NewID();
 531                                 req->error = iter->second->Query(*req);
 532
 533                                 return (req->error.Id() == NO_ERROR) ? sqlsuccess : NULL;
 534                         }
 535                         else
 536                         {
 537                                 req->error.Id(BAD_DBID);
 538                                 return NULL;
 539                         }
 540                 }
 541
 542                 log(DEBUG, "Got unsupported API version string: %s", request->GetId());
 543
 544                 return NULL;
 545         }
 546
 547         virtual void OnUnloadModule(Module* mod, const std::string&     name)
 548         {
 549                 /* When a module unloads we have to check all the pending queries for all our connections
 550                  * and set the Module* specifying where the query came from to NULL. If the query has already
 551                  * been dispatched then when it is processed it will be dropped if the pointer is NULL.
 552                  *
 553                  * If the queries we find are not already being executed then we can simply remove them immediately.
 554                  */
 555                 for(ConnMap::iterator iter = connections.begin(); iter != connections.end(); iter++)
 556                 {
 557                         iter->second->OnUnloadModule(mod);
 558                 }
 559         }
 560
 561         unsigned long NewID()
 562         {
 563                 if (currid+1 == 0)
 564                         currid++;
 565
 566                 return ++currid;
 567         }
 568
 569         virtual Version GetVersion()
 570         {
 571                 return Version(1, 0, 0, 0, VF_VENDOR|VF_SERVICEPROVIDER);
 572         }
 573
 574         virtual ~ModulePgSQL()
 575         {
 576                 DELETE(sqlsuccess);
 577         }
 578 };
 579
 580 SQLConn::SQLConn(ModulePgSQL* self, Server* srv, const std::string &h, unsigned int p, const std::string &d, const std::string &u, const std::string &pwd, bool s)
 581 : InspSocket::InspSocket(), us(self), Srv(srv), dbhost(h), dbport(p), dbname(d), dbuser(u), dbpass(pwd), ssl(s), sql(NULL), status(CWRITE), qinprog(false)
 582 {
 583         log(DEBUG, "Creating new PgSQL connection to database %s on %s:%u (%s/%s)", dbname.c_str(), dbhost.c_str(), dbport, dbuser.c_str(), dbpass.c_str());
 584
 585         /* Some of this could be reviewed, unsure if I need to fill 'host' etc...
 586          * just copied this over from the InspSocket constructor.
 587          */
 588         strlcpy(this->host, dbhost.c_str(), MAXBUF);
 589         this->port = dbport;
 590
 591         this->ClosePending = false;
 592
 593         if(!inet_aton(this->host, &this->addy))
 594         {
 595                 /* Its not an ip, spawn the resolver.
 596                  * PgSQL doesn't do nonblocking DNS
 597                  * lookups, so we do it for it.
 598                  */
 599
 600                 log(DEBUG,"Attempting to resolve %s", this->host);
 601
 602                 this->dns.SetNS(Srv->GetConfig()->DNSServer);
 603                 this->dns.ForwardLookupWithFD(this->host, fd);
 604
 605                 this->state = I_RESOLVING;
 606                 socket_ref[this->fd] = this;
 607
 608                 return;
 609         }
 610         else
 611         {
 612                 log(DEBUG,"No need to resolve %s", this->host);
 613                 strlcpy(this->IP, this->host, MAXBUF);
 614
 615                 if(!this->DoConnect())
 616                 {
 617                         throw ModuleException("Connect failed");
 618                 }
 619         }
 620 }
 621
 622 SQLConn::~SQLConn()
 623 {
 624         Close();
 625 }
 626
 627 bool SQLConn::DoResolve()
 628 {
 629         log(DEBUG, "Checking for DNS lookup result");
 630
 631         if(this->dns.HasResult())
 632         {
 633                 std::string res_ip = dns.GetResultIP();
 634
 635                 if(res_ip.length())
 636                 {
 637                         log(DEBUG, "Got result: %s", res_ip.c_str());
 638
 639                         strlcpy(this->IP, res_ip.c_str(), MAXBUF);
 640                         dbhost = res_ip;
 641
 642                         socket_ref[this->fd] = NULL;
 643
 644                         return this->DoConnect();
 645                 }
 646                 else
 647                 {
 648                         log(DEBUG, "DNS lookup failed, dying horribly");
 649                         Close();
 650                         return false;
 651                 }
 652         }
 653         else
 654         {
 655                 log(DEBUG, "No result for lookup yet!");
 656                 return true;
 657         }
 658 }
 659
 660 bool SQLConn::DoConnect()
 661 {
 662         log(DEBUG, "SQLConn::DoConnect()");
 663
 664         if(!(sql = PQconnectStart(MkInfoStr().c_str())))
 665         {
 666                 log(DEBUG, "Couldn't allocate PGconn structure, aborting: %s", PQerrorMessage(sql));
 667                 Close();
 668                 return false;
 669         }
 670
 671         if(PQstatus(sql) == CONNECTION_BAD)
 672         {
 673                 log(DEBUG, "PQconnectStart failed: %s", PQerrorMessage(sql));
 674                 Close();
 675                 return false;
 676         }
 677
 678         ShowStatus();
 679
 680         if(PQsetnonblocking(sql, 1) == -1)
 681         {
 682                 log(DEBUG, "Couldn't set connection nonblocking: %s", PQerrorMessage(sql));
 683                 Close();
 684                 return false;
 685         }
 686
 687         /* OK, we've initalised the connection, now to get it hooked into the socket engine
 688          * and then start polling it.
 689          */
 690
 691         log(DEBUG, "Old DNS socket: %d", this->fd);
 692         this->fd = PQsocket(sql);
 693         log(DEBUG, "New SQL socket: %d", this->fd);
 694
 695         if(this->fd <= -1)
 696         {
 697                 log(DEBUG, "PQsocket says we have an invalid FD: %d", this->fd);
 698                 Close();
 699                 return false;
 700         }
 701
 702         this->state = I_CONNECTING;
 703         ServerInstance->SE->AddFd(this->fd,false,X_ESTAB_MODULE);
 704         socket_ref[this->fd] = this;
 705
 706         /* Socket all hooked into the engine, now to tell PgSQL to start connecting */
 707
 708         return DoPoll();
 709 }
 710
 711 void SQLConn::Close()
 712 {
 713         log(DEBUG,"SQLConn::Close");
 714
 715         if(this->fd > 01)
 716                 socket_ref[this->fd] = NULL;
 717         this->fd = -1;
 718         this->state = I_ERROR;
 719         this->OnError(I_ERR_SOCKET);
 720         this->ClosePending = true;
 721
 722         if(sql)
 723         {
 724                 PQfinish(sql);
 725                 sql = NULL;
 726         }
 727
 728         return;
 729 }
 730
 731 bool SQLConn::DoPoll()
 732 {
 733         switch(PQconnectPoll(sql))
 734         {
 735                 case PGRES_POLLING_WRITING:
 736                         log(DEBUG, "PGconnectPoll: PGRES_POLLING_WRITING");
 737                         WantWrite();
 738                         status = CWRITE;
 739                         return DoPoll();
 740                 case PGRES_POLLING_READING:
 741                         log(DEBUG, "PGconnectPoll: PGRES_POLLING_READING");
 742                         status = CREAD;
 743                         break;
 744                 case PGRES_POLLING_FAILED:
 745                         log(DEBUG, "PGconnectPoll: PGRES_POLLING_FAILED: %s", PQerrorMessage(sql));
 746                         return false;
 747                 case PGRES_POLLING_OK:
 748                         log(DEBUG, "PGconnectPoll: PGRES_POLLING_OK");
 749                         status = WWRITE;
 750                         return DoConnectedPoll();
 751                 default:
 752                         log(DEBUG, "PGconnectPoll: wtf?");
 753                         break;
 754         }
 755
 756         return true;
 757 }
 758
 759 bool SQLConn::DoConnectedPoll()
 760 {
 761         if(!qinprog && queue.totalsize())
 762         {
 763                 /* There's no query currently in progress, and there's queries in the queue. */
 764                 SQLrequest& query = queue.front();
 765                 DoQuery(query);
 766         }
 767
 768         if(PQconsumeInput(sql))
 769         {
 770                 log(DEBUG, "PQconsumeInput succeeded");
 771
 772                 if(PQisBusy(sql))
 773                 {
 774                         log(DEBUG, "Still busy processing command though");
 775                 }
 776                 else if(qinprog)
 777                 {
 778                         log(DEBUG, "Looks like we have a result to process!");
 779
 780                         /* Grab the request we're processing */
 781                         SQLrequest& query = queue.front();
 782
 783                         log(DEBUG, "ID is %lu", query.id);
 784
 785                         /* Get a pointer to the module we're about to return the result to */
 786                         Module* to = query.GetSource();
 787
 788                         /* Fetch the result.. */
 789                         PGresult* result = PQgetResult(sql);
 790
 791                         /* PgSQL would allow a query string to be sent which has multiple
 792                          * queries in it, this isn't portable across database backends and
 793                          * we don't want modules doing it. But just in case we make sure we
 794                          * drain any results there are and just use the last one.
 795                          * If the module devs are behaving there will only be one result.
 796                          */
 797                         while (PGresult* temp = PQgetResult(sql))
 798                         {
 799                                 PQclear(result);
 800                                 result = temp;
 801                         }
 802
 803                         if(to)
 804                         {
 805                                 /* ..and the result */
 806                                 PgSQLresult reply(us, to, query.id, result);
 807
 808                                 log(DEBUG, "Got result, status code: %s; error message: %s", PQresStatus(PQresultStatus(result)), PQresultErrorMessage(result));
 809
 810                                 switch(PQresultStatus(result))
 811                                 {
 812                                         case PGRES_EMPTY_QUERY:
 813                                         case PGRES_BAD_RESPONSE:
 814                                         case PGRES_FATAL_ERROR:
 815                                                 reply.error.Id(QREPLY_FAIL);
 816                                                 reply.error.Str(PQresultErrorMessage(result));
 817                                         default:;
 818                                                 /* No action, other values are not errors */
 819                                 }
 820
 821                                 reply.Send();
 822
 823                                 /* PgSQLresult's destructor will free the PGresult */
 824                         }
 825                         else
 826                         {
 827                                 /* If the client module is unloaded partway through a query then the provider will set
 828                                  * the pointer to NULL. We cannot just cancel the query as the result will still come
 829                                  * through at some point...and it could get messy if we play with invalid pointers...
 830                                  */
 831                                 log(DEBUG, "Looks like we're handling a zombie query from a module which unloaded before it got a result..fun. ID: %lu", query.id);
 832                                 PQclear(result);
 833                         }
 834
 835                         qinprog = false;
 836                         queue.pop();
 837                         DoConnectedPoll();
 838                 }
 839
 840                 return true;
 841         }
 842
 843         log(DEBUG, "PQconsumeInput failed: %s", PQerrorMessage(sql));
 844         return false;
 845 }
 846
 847 void SQLConn::ShowStatus()
 848 {
 849         switch(PQstatus(sql))
 850         {
 851                 case CONNECTION_STARTED:
 852                         log(DEBUG, "PQstatus: CONNECTION_STARTED: Waiting for connection to be made.");
 853                         break;
 854
 855                 case CONNECTION_MADE:
 856                         log(DEBUG, "PQstatus: CONNECTION_MADE: Connection OK; waiting to send.");
 857                         break;
 858
 859                 case CONNECTION_AWAITING_RESPONSE:
 860                         log(DEBUG, "PQstatus: CONNECTION_AWAITING_RESPONSE: Waiting for a response from the server.");
 861                         break;
 862
 863                 case CONNECTION_AUTH_OK:
 864                         log(DEBUG, "PQstatus: CONNECTION_AUTH_OK: Received authentication; waiting for backend start-up to finish.");
 865                         break;
 866
 867                 case CONNECTION_SSL_STARTUP:
 868                         log(DEBUG, "PQstatus: CONNECTION_SSL_STARTUP: Negotiating SSL encryption.");
 869                         break;
 870
 871                 case CONNECTION_SETENV:
 872                         log(DEBUG, "PQstatus: CONNECTION_SETENV: Negotiating environment-driven parameter settings.");
 873                         break;
 874
 875                 default:
 876                         log(DEBUG, "PQstatus: ???");
 877         }
 878 }
 879
 880 bool SQLConn::OnDataReady()
 881 {
 882         /* Always return true here, false would close the socket - we need to do that ourselves with the pgsql API */
 883         log(DEBUG, "OnDataReady(): status = %s", StatusStr());
 884
 885         return DoEvent();
 886 }
 887
 888 bool SQLConn::OnWriteReady()
 889 {
 890         /* Always return true here, false would close the socket - we need to do that ourselves with the pgsql API */
 891         log(DEBUG, "OnWriteReady(): status = %s", StatusStr());
 892
 893         return DoEvent();
 894 }
 895
 896 bool SQLConn::OnConnected()
 897 {
 898         log(DEBUG, "OnConnected(): status = %s", StatusStr());
 899
 900         return DoEvent();
 901 }
 902
 903 bool SQLConn::DoEvent()
 904 {
 905         bool ret;
 906
 907         if((status == CREAD) || (status == CWRITE))
 908         {
 909                 ret = DoPoll();
 910         }
 911         else
 912         {
 913                 ret = DoConnectedPoll();
 914         }
 915
 916         switch(PQflush(sql))
 917         {
 918                 case -1:
 919                         log(DEBUG, "Error flushing write queue: %s", PQerrorMessage(sql));
 920                         break;
 921                 case 0:
 922                         log(DEBUG, "Successfully flushed write queue (or there was nothing to write)");
 923                         break;
 924                 case 1:
 925                         log(DEBUG, "Not all of the write queue written, triggering write event so we can have another go");
 926                         WantWrite();
 927                         break;
 928         }
 929
 930         return ret;
 931 }
 932
 933 std::string SQLConn::MkInfoStr()
 934 {
 935         std::ostringstream conninfo("connect_timeout = '2'");
 936
 937         if(dbhost.length())
 938                 conninfo << " hostaddr = '" << dbhost << "'";
 939
 940         if(dbport)
 941                 conninfo << " port = '" << dbport << "'";
 942
 943         if(dbname.length())
 944                 conninfo << " dbname = '" << dbname << "'";
 945
 946         if(dbuser.length())
 947                 conninfo << " user = '" << dbuser << "'";
 948
 949         if(dbpass.length())
 950                 conninfo << " password = '" << dbpass << "'";
 951
 952         if(ssl)
 953                 conninfo << " sslmode = 'require'";
 954
 955         return conninfo.str();
 956 }
 957
 958 const char* SQLConn::StatusStr()
 959 {
 960         if(status == CREAD) return "CREAD";
 961         if(status == CWRITE) return "CWRITE";
 962         if(status == WREAD) return "WREAD";
 963         if(status == WWRITE) return "WWRITE";
 964         return "Err...what, erm..BUG!";
 965 }
 966
 967 SQLerror SQLConn::DoQuery(SQLrequest &req)
 968 {
 969         if((status == WREAD) || (status == WWRITE))
 970         {
 971                 if(!qinprog)
 972                 {
 973                         /* Parse the command string and dispatch it */
 974
 975                         /* Pointer to the buffer we screw around with substitution in */
 976                         char* query;
 977                         /* Pointer to the current end of query, where we append new stuff */
 978                         char* queryend;
 979                         /* Total length of the unescaped parameters */
 980                         unsigned int paramlen;
 981
 982                         paramlen = 0;
 983
 984                         for(ParamL::iterator i = req.query.p.begin(); i != req.query.p.end(); i++)
 985                         {
 986                                 paramlen += i->size();
 987                         }
 988
 989                         /* To avoid a lot of allocations, allocate enough memory for the biggest the escaped query could possibly be.
 990                          * sizeofquery + (totalparamlength*2) + 1
 991                          *
 992                          * The +1 is for null-terminating the string for PQsendQuery()
 993                          */
 994
 995                         query = new char[req.query.q.length() + (paramlen*2)];
 996                         queryend = query;
 997
 998                         /* Okay, now we have a buffer large enough we need to start copying the query into it and escaping and substituting
 999                          * the parameters into it...
1000                          */
1001
1002                         for(unsigned int i = 0; i < req.query.q.length(); i++)
1003                         {
1004                                 if(req.query.q[i] == '?')
1005                                 {
1006                                         /* We found a place to substitute..what fun.
1007                                          * Use the PgSQL calls to escape and write the
1008                                          * escaped string onto the end of our query buffer,
1009                                          * then we "just" need to make sure queryend is
1010                                          * pointing at the right place.
1011                                          */
1012
1013                                         if(req.query.p.size())
1014                                         {
1015                                                 int error = 0;
1016                                                 size_t len = 0;
1017
1018 #ifdef PGSQL_HAS_ESCAPECONN
1019                                                 len = PQescapeStringConn(sql, queryend, req.query.p.front().c_str(), req.query.p.front().length(), &error);
1020 #else
1021                                                 len = PQescapeStringConn(queryend, req.query.p.front().c_str(), req.query.p.front().length());
1022                                                 error = 0;
1023 #endif
1024
1025                                                 if(error)
1026                                                 {
1027                                                         log(DEBUG, "Apparently PQescapeStringConn() failed somehow...don't know how or what to do...");
1028                                                 }
1029
1030                                                 log(DEBUG, "Appended %d bytes of escaped string onto the query", len);
1031
1032                                                 /* Incremenet queryend to the end of the newly escaped parameter */
1033                                                 queryend += len;
1034
1035                                                 /* Remove the parameter we just substituted in */
1036                                                 req.query.p.pop_front();
1037                                         }
1038                                         else
1039                                         {
1040                                                 log(DEBUG, "Found a substitution location but no parameter to substitute :|");
1041                                                 break;
1042                                         }
1043                                 }
1044                                 else
1045                                 {
1046                                         *queryend = req.query.q[i];
1047                                         queryend++;
1048                                 }
1049                         }
1050
1051                         /* Null-terminate the query */
1052                         *queryend = 0;
1053
1054                         log(DEBUG, "Attempting to dispatch query: %s", query);
1055
1056                         req.query.q = query;
1057
1058                         if(PQsendQuery(sql, query))
1059                         {
1060                                 log(DEBUG, "Dispatched query successfully");
1061                                 qinprog = true;
1062                                 DELETE(query);
1063                                 return SQLerror();
1064                         }
1065                         else
1066                         {
1067                                 log(DEBUG, "Failed to dispatch query: %s", PQerrorMessage(sql));
1068                                 DELETE(query);
1069                                 return SQLerror(QSEND_FAIL, PQerrorMessage(sql));
1070                         }
1071                 }
1072         }
1073
1074         log(DEBUG, "Can't query until connection is complete");
1075         return SQLerror(BAD_CONN, "Can't query until connection is complete");
1076 }
1077
1078 SQLerror SQLConn::Query(const SQLrequest &req)
1079 {
1080         queue.push(req);
1081
1082         if(!qinprog && queue.totalsize())
1083         {
1084                 /* There's no query currently in progress, and there's queries in the queue. */
1085                 SQLrequest& query = queue.front();
1086                 return DoQuery(query);
1087         }
1088         else
1089         {
1090                 return SQLerror();
1091         }
1092 }
1093
1094 void SQLConn::OnUnloadModule(Module* mod)
1095 {
1096         queue.PurgeModule(mod);
1097 }
1098
1099 class ModulePgSQLFactory : public ModuleFactory
1100 {
1101  public:
1102         ModulePgSQLFactory()
1103         {
1104         }
1105
1106         ~ModulePgSQLFactory()
1107         {
1108         }
1109
1110         virtual Module * CreateModule(Server* Me)
1111         {
1112                 return new ModulePgSQL(Me);
1113         }
1114 };
1115
1116
1117 extern "C" void * init_module( void )
1118 {
1119         return new ModulePgSQLFactory;
1120 }