src/modules/extra/m_pgsql.cpp

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