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 <sstream>
  19 #include <string>
  20 #include <deque>
  21 #include <map>
  22 #include <libpq-fe.h>
  23
  24 #include "users.h"
  25 #include "channels.h"
  26 #include "modules.h"
  27 #include "helperfuncs.h"
  28 #include "inspircd.h"
  29 #include "configreader.h"
  30
  31 #include "m_sqlv2.h"
  32
  33 /* $ModDesc: PostgreSQL Service Provider module for all other m_sql* modules, uses v2 of the SQL API */
  34 /* $CompileFlags: -I`pg_config --includedir` */
  35 /* $LinkerFlags: -L`pg_config --libdir` -lpq */
  36
  37 /* UGH, UGH, UGH, UGH, UGH, UGH
  38  * I'm having trouble seeing how I
  39  * can avoid this. The core-defined
  40  * constructors for InspSocket just
  41  * aren't suitable...and if I'm
  42  * reimplementing them I need this so
  43  * I can access the socket engine :\
  44  */
  45 extern InspIRCd* ServerInstance;
  46 InspSocket* socket_ref[MAX_DESCRIPTORS];
  47
  48 /* Forward declare, so we can have the typedef neatly at the top */
  49 class SQLConn;
  50
  51 typedef std::map<std::string, SQLConn*> ConnMap;
  52
  53 /* CREAD,       Connecting and wants read event
  54  * CWRITE,      Connecting and wants write event
  55  * WREAD,       Connected/Working and wants read event
  56  * WWRITE,      Connected/Working and wants write event
  57  */
  58 enum SQLstatus { CREAD, CWRITE, WREAD, WWRITE };
  59
  60 /** QueryQueue, a queue of queries waiting to be executed.
  61  * This maintains two queues internally, one for 'priority'
  62  * queries and one for less important ones. Each queue has
  63  * new queries appended to it and ones to execute are popped
  64  * off the front. This keeps them flowing round nicely and no
  65  * query should ever get 'stuck' for too long. If there are
  66  * queries in the priority queue they will be executed first,
  67  * 'unimportant' queries will only be executed when the
  68  * priority queue is empty.
  69  *
  70  * These are lists of SQLresult so we can, from the moment the
  71  * SQLrequest is recieved, be beginning to construct the result
  72  * object. The copy in the deque can then be submitted in-situ
  73  * and finally deleted from this queue. No copies of the SQLresult :)
  74  *
  75  * Because we work on the SQLresult in-situ, we need a way of accessing the
  76  * result we are currently processing, QueryQueue::front(), but that call
  77  * needs to always return the same element until that element is removed
  78  * from the queue, this is what the 'which' variable is. New queries are
  79  * always added to the back of one of the two queues, but if when front()
  80  * is first called then the priority queue is empty then front() will return
  81  * a query from the normal queue, but if a query is then added to the priority
  82  * queue then front() must continue to return the front of the *normal* queue
  83  * until pop() is called.
  84  */
  85
  86 class QueryQueue : public classbase
  87 {
  88 private:
  89         std::deque<SQLresult> priority; /* The priority queue */
  90         std::deque<SQLresult> normal;   /* The 'normal' queue */
  91         enum { PRI, NOR, NON } which;   /* Which queue the currently active element is at the front of */
  92
  93 public:
  94         QueryQueue()
  95         : which(NON)
  96         {
  97
  98         }
  99
 100         void push(const Query &q, bool pri = false)
 101         {
 102                 log(DEBUG, "QueryQueue::push_back(): Adding %s query to queue: %s", ((pri) ? "priority" : "non-priority"), q.c_str());
 103
 104                 if(pri)
 105                         priority.push_back(q);
 106                 else
 107                         normal.push_back(q);
 108         }
 109
 110         void pop()
 111         {
 112                 if((which == PRI) && priority.size())
 113                 {
 114                         priority.pop_front();
 115                 }
 116                 else if((which == NOR) && normal.size())
 117                 {
 118                         normal.pop_front();
 119                 }
 120
 121                 /* Silently do nothing if there was no element to pop() */
 122         }
 123
 124         SQLresult& front()
 125         {
 126                 switch(which)
 127                 {
 128                         case PRI:
 129                                 return priority.front();
 130                         case NOR:
 131                                 return normal.front();
 132                         default:
 133                                 if(priority.size())
 134                                 {
 135                                         which = PRI;
 136                                         return priority.front();
 137                                 }
 138
 139                                 if(normal.size())
 140                                 {
 141                                         which = NOR;
 142                                         return normal.front();
 143                                 }
 144
 145                                 /* This will probably result in a segfault,
 146                                  * but the caller should have checked totalsize()
 147                                  * first so..meh - moron :p
 148                                  */
 149
 150                                 return priority.front();
 151                 }
 152         }
 153
 154         std::pair<int, int> size()
 155         {
 156                 return std::make_pair(priority.size(), normal.size());
 157         }
 158
 159         int totalsize()
 160         {
 161                 return priority.size() + normal.size();
 162         }
 163 };
 164
 165 /** SQLConn represents one SQL session.
 166  * Each session has its own persistent connection to the database.
 167  * This is a subclass of InspSocket so it can easily recieve read/write events from the core socket
 168  * engine, unlike the original MySQL module this module does not block. Ever. It gets a mild stabbing
 169  * if it dares to.
 170  */
 171
 172 class SQLConn : public InspSocket
 173 {
 174 private:
 175         Server* Srv;                    /* Server* for..uhm..something, maybe */
 176         std::string     dbhost; /* Database server hostname */
 177         unsigned int    dbport; /* Database server port */
 178         std::string     dbname; /* Database name */
 179         std::string     dbuser; /* Database username */
 180         std::string     dbpass; /* Database password */
 181         bool                    ssl;    /* If we should require SSL */
 182         PGconn*                 sql;    /* PgSQL database connection handle */
 183         SQLstatus               status; /* PgSQL database connection status */
 184         bool                    qinprog;/* If there is currently a query in progress */
 185         QueryQueue              queue;  /* Queue of queries waiting to be executed on this connection */
 186         Query                   query;  /* The currently active query on this connection */
 187
 188 public:
 189
 190         /* This class should only ever be created inside this module, using this constructor, so we don't have to worry about the default ones */
 191
 192         SQLConn(Server* srv, const std::string &h, unsigned int p, const std::string &d, const std::string &u, const std::string &pwd, bool s)
 193         : InspSocket::InspSocket(), Srv(srv), dbhost(h), dbport(p), dbname(d), dbuser(u), dbpass(pwd), ssl(s), sql(NULL), status(CWRITE), qinprog(false)
 194         {
 195                 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());
 196
 197                 /* Some of this could be reviewed, unsure if I need to fill 'host' etc...
 198                  * just copied this over from the InspSocket constructor.
 199                  */
 200                 strlcpy(this->host, dbhost.c_str(), MAXBUF);
 201                 this->port = dbport;
 202
 203                 this->ClosePending = false;
 204
 205                 if(!inet_aton(this->host, &this->addy))
 206                 {
 207                         /* Its not an ip, spawn the resolver.
 208                          * PgSQL doesn't do nonblocking DNS
 209                          * lookups, so we do it for it.
 210                          */
 211
 212                         log(DEBUG,"Attempting to resolve %s", this->host);
 213
 214                         this->dns.SetNS(Srv->GetConfig()->DNSServer);
 215                         this->dns.ForwardLookupWithFD(this->host, fd);
 216
 217                         this->state = I_RESOLVING;
 218                         socket_ref[this->fd] = this;
 219
 220                         return;
 221                 }
 222                 else
 223                 {
 224                         log(DEBUG,"No need to resolve %s", this->host);
 225                         strlcpy(this->IP, this->host, MAXBUF);
 226
 227                         if(!this->DoConnect())
 228                         {
 229                                 throw ModuleException("Connect failed");
 230                         }
 231                 }
 232         }
 233
 234         ~SQLConn()
 235         {
 236                 Close();
 237         }
 238
 239         bool DoResolve()
 240         {
 241                 log(DEBUG, "Checking for DNS lookup result");
 242
 243                 if(this->dns.HasResult())
 244                 {
 245                         std::string res_ip = dns.GetResultIP();
 246
 247                         if(res_ip.length())
 248                         {
 249                                 log(DEBUG, "Got result: %s", res_ip.c_str());
 250
 251                                 strlcpy(this->IP, res_ip.c_str(), MAXBUF);
 252                                 dbhost = res_ip;
 253
 254                                 socket_ref[this->fd] = NULL;
 255
 256                                 return this->DoConnect();
 257                         }
 258                         else
 259                         {
 260                                 log(DEBUG, "DNS lookup failed, dying horribly");
 261                                 Close();
 262                                 return false;
 263                         }
 264                 }
 265                 else
 266                 {
 267                         log(DEBUG, "No result for lookup yet!");
 268                         return true;
 269                 }
 270         }
 271
 272         bool DoConnect()
 273         {
 274                 log(DEBUG, "SQLConn::DoConnect()");
 275
 276                 if(!(sql = PQconnectStart(MkInfoStr().c_str())))
 277                 {
 278                         log(DEBUG, "Couldn't allocate PGconn structure, aborting: %s", PQerrorMessage(sql));
 279                         Close();
 280                         return false;
 281                 }
 282
 283                 if(PQstatus(sql) == CONNECTION_BAD)
 284                 {
 285                         log(DEBUG, "PQconnectStart failed: %s", PQerrorMessage(sql));
 286                         Close();
 287                         return false;
 288                 }
 289
 290                 ShowStatus();
 291
 292                 if(PQsetnonblocking(sql, 1) == -1)
 293                 {
 294                         log(DEBUG, "Couldn't set connection nonblocking: %s", PQerrorMessage(sql));
 295                         Close();
 296                         return false;
 297                 }
 298
 299                 /* OK, we've initalised the connection, now to get it hooked into the socket engine
 300                  * and then start polling it.
 301                  */
 302
 303                 log(DEBUG, "Old DNS socket: %d", this->fd);
 304                 this->fd = PQsocket(sql);
 305                 log(DEBUG, "New SQL socket: %d", this->fd);
 306
 307                 if(this->fd <= -1)
 308                 {
 309                         log(DEBUG, "PQsocket says we have an invalid FD: %d", this->fd);
 310                         Close();
 311                         return false;
 312                 }
 313
 314                 this->state = I_CONNECTING;
 315                 ServerInstance->SE->AddFd(this->fd,false,X_ESTAB_MODULE);
 316                 socket_ref[this->fd] = this;
 317
 318                 /* Socket all hooked into the engine, now to tell PgSQL to start connecting */
 319
 320                 return DoPoll();
 321         }
 322
 323         virtual void Close()
 324         {
 325                 log(DEBUG,"SQLConn::Close");
 326
 327                 if(this->fd > 01)
 328                         socket_ref[this->fd] = NULL;
 329                 this->fd = -1;
 330                 this->state = I_ERROR;
 331                 this->OnError(I_ERR_SOCKET);
 332                 this->ClosePending = true;
 333
 334                 if(sql)
 335                 {
 336                         PQfinish(sql);
 337                         sql = NULL;
 338                 }
 339
 340                 return;
 341         }
 342
 343         bool DoPoll()
 344         {
 345                 switch(PQconnectPoll(sql))
 346                 {
 347                         case PGRES_POLLING_WRITING:
 348                                 log(DEBUG, "PGconnectPoll: PGRES_POLLING_WRITING");
 349                                 WantWrite();
 350                                 status = CWRITE;
 351                                 return DoPoll();
 352                         case PGRES_POLLING_READING:
 353                                 log(DEBUG, "PGconnectPoll: PGRES_POLLING_READING");
 354                                 status = CREAD;
 355                                 break;
 356                         case PGRES_POLLING_FAILED:
 357                                 log(DEBUG, "PGconnectPoll: PGRES_POLLING_FAILED: %s", PQerrorMessage(sql));
 358                                 return false;
 359                         case PGRES_POLLING_OK:
 360                                 log(DEBUG, "PGconnectPoll: PGRES_POLLING_OK");
 361                                 status = WWRITE;
 362                                 return DoConnectedPoll()
 363                         default:
 364                                 log(DEBUG, "PGconnectPoll: wtf?");
 365                                 break;
 366                 }
 367
 368                 return true;
 369         }
 370
 371         bool DoConnectedPoll()
 372         {
 373                 if(!qinprog && queue.totalsize())
 374                 {
 375                         /* There's no query currently in progress, and there's queries in the queue. */
 376                         query = queue.pop_front();
 377                         DoQuery();
 378                 }
 379
 380                 if(PQconsumeInput(sql))
 381                 {
 382                         log(DEBUG, "PQconsumeInput succeeded");
 383
 384                         if(PQisBusy(sql))
 385                         {
 386                                 log(DEBUG, "Still busy processing command though");
 387                         }
 388                         else
 389                         {
 390                                 log(DEBUG, "Looks like we have a result to process!");
 391
 392                                 while(PGresult* result = PQgetResult(sql))
 393                                 {
 394                                         int cols = PQnfields(result);
 395
 396                                         log(DEBUG, "Got result! :D");
 397                                         log(DEBUG, "%d rows, %d columns checking now what the column names are", PQntuples(result), cols);
 398
 399                                         for(int i = 0; i < cols; i++)
 400                                         {
 401                                                 log(DEBUG, "Column name: %s (%d)", PQfname(result, i));
 402                                         }
 403
 404                                         PQclear(result);
 405                                 }
 406
 407                                 qinprog = false;
 408                         }
 409
 410                         return true;
 411                 }
 412
 413                 log(DEBUG, "PQconsumeInput failed: %s", PQerrorMessage(sql));
 414                 return false;
 415         }
 416
 417         void ShowStatus()
 418         {
 419                 switch(PQstatus(sql))
 420                 {
 421                         case CONNECTION_STARTED:
 422                                 log(DEBUG, "PQstatus: CONNECTION_STARTED: Waiting for connection to be made.");
 423                                 break;
 424
 425                         case CONNECTION_MADE:
 426                                 log(DEBUG, "PQstatus: CONNECTION_MADE: Connection OK; waiting to send.");
 427                                 break;
 428
 429                         case CONNECTION_AWAITING_RESPONSE:
 430                                 log(DEBUG, "PQstatus: CONNECTION_AWAITING_RESPONSE: Waiting for a response from the server.");
 431                                 break;
 432
 433                         case CONNECTION_AUTH_OK:
 434                                 log(DEBUG, "PQstatus: CONNECTION_AUTH_OK: Received authentication; waiting for backend start-up to finish.");
 435                                 break;
 436
 437                         case CONNECTION_SSL_STARTUP:
 438                                 log(DEBUG, "PQstatus: CONNECTION_SSL_STARTUP: Negotiating SSL encryption.");
 439                                 break;
 440
 441                         case CONNECTION_SETENV:
 442                                 log(DEBUG, "PQstatus: CONNECTION_SETENV: Negotiating environment-driven parameter settings.");
 443                                 break;
 444
 445                         default:
 446                                 log(DEBUG, "PQstatus: ???");
 447                 }
 448         }
 449
 450         virtual bool OnDataReady()
 451         {
 452                 /* Always return true here, false would close the socket - we need to do that ourselves with the pgsql API */
 453                 log(DEBUG, "OnDataReady(): status = %s", StatusStr());
 454
 455                 return DoEvent();
 456         }
 457
 458         virtual bool OnWriteReady()
 459         {
 460                 /* Always return true here, false would close the socket - we need to do that ourselves with the pgsql API */
 461                 log(DEBUG, "OnWriteReady(): status = %s", StatusStr());
 462
 463                 return DoEvent();
 464         }
 465
 466         virtual bool OnConnected()
 467         {
 468                 log(DEBUG, "OnConnected(): status = %s", StatusStr());
 469
 470                 return DoEvent();
 471         }
 472
 473         bool DoEvent()
 474         {
 475                 bool ret;
 476
 477                 if((status == CREAD) || (status == CWRITE))
 478                 {
 479                         ret = DoPoll();
 480                 }
 481                 else
 482                 {
 483                         ret = DoConnectedPoll();
 484                 }
 485
 486                 switch(PQflush(sql))
 487                 {
 488                         case -1:
 489                                 log(DEBUG, "Error flushing write queue: %s", PQerrorMessage(sql));
 490                                 break;
 491                         case 0:
 492                                 log(DEBUG, "Successfully flushed write queue (or there was nothing to write)");
 493                                 break;
 494                         case 1:
 495                                 log(DEBUG, "Not all of the write queue written, triggering write event so we can have another go");
 496                                 WantWrite();
 497                                 break;
 498                 }
 499
 500                 return ret;
 501         }
 502
 503         std::string MkInfoStr()
 504         {
 505                 /* XXX - This needs nonblocking DNS lookups */
 506
 507                 std::ostringstream conninfo("connect_timeout = '2'");
 508
 509                 if(dbhost.length())
 510                         conninfo << " hostaddr = '" << dbhost << "'";
 511
 512                 if(dbport)
 513                         conninfo << " port = '" << dbport << "'";
 514
 515                 if(dbname.length())
 516                         conninfo << " dbname = '" << dbname << "'";
 517
 518                 if(dbuser.length())
 519                         conninfo << " user = '" << dbuser << "'";
 520
 521                 if(dbpass.length())
 522                         conninfo << " password = '" << dbpass << "'";
 523
 524                 if(ssl)
 525                         conninfo << " sslmode = 'require'";
 526
 527                 return conninfo.str();
 528         }
 529
 530         const char* StatusStr()
 531         {
 532                 if(status == CREAD) return "CREAD";
 533                 if(status == CWRITE) return "CWRITE";
 534                 if(status == WREAD) return "WREAD";
 535                 if(status == WWRITE) return "WWRITE";
 536                 return "Err...what, erm..BUG!";
 537         }
 538
 539         SQLerror Query(const Query &query, bool pri)
 540         {
 541                 queue.push_back(query, pri);
 542
 543                 if((status == WREAD) || (status == WWRITE))
 544                 {
 545                         if(!qinprog)
 546                         {
 547                                 if(PQsendQuery(sql, query.c_str()))
 548                                 {
 549                                         log(DEBUG, "Dispatched query: %s", query.c_str());
 550                                         qinprog = true;
 551                                         return SQLerror();
 552                                 }
 553                                 else
 554                                 {
 555                                         log(DEBUG, "Failed to dispatch query: %s", PQerrorMessage(sql));
 556                                         return SQLerror(QSEND_FAIL, PQerrorMessage(sql));
 557                                 }
 558                         }
 559                 }
 560
 561                 log(DEBUG, "Can't query until connection is complete");
 562                 return SQLerror(BAD_CONN, "Can't query until connection is complete");
 563         }
 564 };
 565
 566 class ModulePgSQL : public Module
 567 {
 568 private:
 569         Server* Srv;
 570         ConnMap connections;
 571
 572 public:
 573         ModulePgSQL(Server* Me)
 574         : Module::Module(Me), Srv(Me)
 575         {
 576                 log(DEBUG, "%s 'SQL' feature", Srv->PublishFeature("SQL", this) ? "Published" : "Couldn't publish");
 577                 log(DEBUG, "%s 'PgSQL' feature", Srv->PublishFeature("PgSQL", this) ? "Published" : "Couldn't publish");
 578
 579                 OnRehash("");
 580         }
 581
 582         void Implements(char* List)
 583         {
 584                 List[I_OnRequest] = List[I_OnRehash] = List[I_OnUserRegister] = List[I_OnCheckReady] = List[I_OnUserDisconnect] = 1;
 585         }
 586
 587         virtual void OnRehash(const std::string &parameter)
 588         {
 589                 ConfigReader conf;
 590
 591                 /* Delete all the SQLConn objects in the connection lists,
 592                  * this will call their destructors where they can handle
 593                  * closing connections and such.
 594                  */
 595                 for(ConnMap::iterator iter = connections.begin(); iter != connections.end(); iter++)
 596                 {
 597                         DELETE(iter->second);
 598                 }
 599
 600                 /* Empty out our list of connections */
 601                 connections.clear();
 602
 603                 for(int i = 0; i < conf.Enumerate("database"); i++)
 604                 {
 605                         std::string id;
 606                         SQLConn* newconn;
 607
 608                         id = conf.ReadValue("database", "id", i);
 609                         newconn = new SQLConn(Srv,      conf.ReadValue("database", "hostname", i),
 610                                                                                 conf.ReadInteger("database", "port", i, true),
 611                                                                                 conf.ReadValue("database", "name", i),
 612                                                                                 conf.ReadValue("database", "username", i),
 613                                                                                 conf.ReadValue("database", "password", i),
 614                                                                                 conf.ReadFlag("database", "ssl", i));
 615
 616                         connections.insert(std::make_pair(id, newconn));
 617                 }
 618         }
 619
 620         virtual char* OnRequest(Request* request)
 621         {
 622                 if(strcmp(SQLREQID, request->GetData()) == 0)
 623                 {
 624                         SQLrequest* req = (SQLrequest*)request;
 625                         ConnMap::iterator iter;
 626
 627                         log(DEBUG, "Got query: '%s' on id '%s'", req->query.c_str(), req->dbid.c_str());
 628
 629                         if((iter = connections.find(req->dbid)) != connections.end())
 630                         {
 631                                 /* Execute query */
 632                                 req->error = iter->second->Query(Query(req->query, req->GetSource(), this), req->pri);
 633
 634                                 return SQLSUCCESS;
 635                         }
 636                         else
 637                         {
 638                                 req->error.Id(BAD_DBID);
 639                                 return NULL;
 640                         }
 641                 }
 642
 643                 log(DEBUG, "Got unsupported API version string: %s", request->GetData());
 644
 645                 return NULL;
 646         }
 647
 648         virtual Version GetVersion()
 649         {
 650                 return Version(1, 0, 0, 0, VF_VENDOR|VF_SERVICEPROVIDER);
 651         }
 652
 653         virtual ~ModulePgSQL()
 654         {
 655         }
 656 };
 657
 658 class ModulePgSQLFactory : public ModuleFactory
 659 {
 660  public:
 661         ModulePgSQLFactory()
 662         {
 663         }
 664
 665         ~ModulePgSQLFactory()
 666         {
 667         }
 668
 669         virtual Module * CreateModule(Server* Me)
 670         {
 671                 return new ModulePgSQL(Me);
 672         }
 673 };
 674
 675
 676 extern "C" void * init_module( void )
 677 {
 678         return new ModulePgSQLFactory;
 679 }