src/modules/extra/m_pgsql.cpp

   1 /*       +------------------------------------+
   2  *       | Inspire Internet Relay Chat Daemon |
   3  *       +------------------------------------+
   4  *
   5  *  InspIRCd: (C) 2002-2010 InspIRCd Development Team
   6  * See: http://wiki.inspircd.org/Credits
   7  *
   8  * This program is free but copyrighted software; see
   9  *            the file COPYING for details.
  10  *
  11  * ---------------------------------------------------
  12  */
  13
  14 #include "inspircd.h"
  15 #include <cstdlib>
  16 #include <sstream>
  17 #include <libpq-fe.h>
  18 #include "sql.h"
  19
  20 /* $ModDesc: PostgreSQL Service Provider module for all other m_sql* modules, uses v2 of the SQL API */
  21 /* $CompileFlags: -Iexec("pg_config --includedir") eval("my $s = `pg_config --version`;$s =~ /^.*?(\d+)\.(\d+)\.(\d+).*?$/;my $v = hex(sprintf("0x%02x%02x%02x", $1, $2, $3));print "-DPGSQL_HAS_ESCAPECONN" if(($v >= 0x080104) || ($v >= 0x07030F && $v < 0x070400) || ($v >= 0x07040D && $v < 0x080000) || ($v >= 0x080008 && $v < 0x080100));") */
  22 /* $LinkerFlags: -Lexec("pg_config --libdir") -lpq */
  23 /* $ModDep: m_sqlv2.h */
  24
  25 /* SQLConn rewritten by peavey to
  26  * use EventHandler instead of
  27  * BufferedSocket. This is much neater
  28  * and gives total control of destroy
  29  * and delete of resources.
  30  */
  31
  32 /* Forward declare, so we can have the typedef neatly at the top */
  33 class SQLConn;
  34 class ModulePgSQL;
  35
  36 typedef std::map<std::string, SQLConn*> ConnMap;
  37
  38 /* CREAD,       Connecting and wants read event
  39  * CWRITE,      Connecting and wants write event
  40  * WREAD,       Connected/Working and wants read event
  41  * WWRITE,      Connected/Working and wants write event
  42  * RREAD,       Resetting and wants read event
  43  * RWRITE,      Resetting and wants write event
  44  */
  45 enum SQLstatus { CREAD, CWRITE, WREAD, WWRITE, RREAD, RWRITE };
  46
  47 class ReconnectTimer : public Timer
  48 {
  49  private:
  50         ModulePgSQL* mod;
  51  public:
  52         ReconnectTimer(ModulePgSQL* m) : Timer(5, ServerInstance->Time(), false), mod(m)
  53         {
  54         }
  55         virtual void Tick(time_t TIME);
  56 };
  57
  58
  59 /** PgSQLresult is a subclass of the mostly-pure-virtual class SQLresult.
  60  * All SQL providers must create their own subclass and define it's methods using that
  61  * database library's data retriveal functions. The aim is to avoid a slow and inefficient process
  62  * of converting all data to a common format before it reaches the result structure. This way
  63  * data is passes to the module nearly as directly as if it was using the API directly itself.
  64  */
  65
  66 class PgSQLresult : public SQLResult
  67 {
  68         PGresult* res;
  69         int currentrow;
  70         int rows;
  71  public:
  72         PgSQLresult(PGresult* result) : res(result), currentrow(0)
  73         {
  74                 rows = PQntuples(res);
  75                 if (!rows)
  76                         rows = atoi(PQcmdTuples(res));
  77         }
  78
  79         ~PgSQLresult()
  80         {
  81                 PQclear(res);
  82         }
  83
  84         virtual int Rows()
  85         {
  86                 return rows;
  87         }
  88
  89         virtual void GetCols(std::vector<std::string>& result)
  90         {
  91                 result.resize(PQnfields(res));
  92                 for(unsigned int i=0; i < result.size(); i++)
  93                 {
  94                         result[i] = PQfname(res, i);
  95                 }
  96         }
  97
  98         virtual SQLEntry GetValue(int row, int column)
  99         {
 100                 char* v = PQgetvalue(res, row, column);
 101                 if (!v || PQgetisnull(res, row, column))
 102                         return SQLEntry();
 103
 104                 return SQLEntry(std::string(v, PQgetlength(res, row, column)));
 105         }
 106
 107         virtual bool GetRow(SQLEntries& result)
 108         {
 109                 if (currentrow >= PQntuples(res))
 110                         return false;
 111                 int ncols = PQnfields(res);
 112
 113                 for(int i = 0; i < ncols; i++)
 114                 {
 115                         result.push_back(GetValue(currentrow, i));
 116                 }
 117                 currentrow++;
 118
 119                 return true;
 120         }
 121 };
 122
 123 /** SQLConn represents one SQL session.
 124  */
 125 class SQLConn : public SQLProvider, public EventHandler
 126 {
 127  private:
 128         reference<ConfigTag> conf;      /* The <database> entry */
 129         std::deque<SQLQuery*> queue;
 130         PGconn*                 sql;            /* PgSQL database connection handle */
 131         SQLstatus               status;         /* PgSQL database connection status */
 132         SQLQuery*               qinprog;        /* If there is currently a query in progress */
 133         time_t                  idle;           /* Time we last heard from the database */
 134
 135  public:
 136         SQLConn(Module* Creator, ConfigTag* tag)
 137         : SQLProvider(Creator, "SQL/" + tag->getString("id")), conf(tag), sql(NULL), status(CWRITE), qinprog(NULL)
 138         {
 139                 idle = ServerInstance->Time();
 140                 if (!DoConnect())
 141                 {
 142                         ServerInstance->Logs->Log("m_pgsql",DEFAULT, "WARNING: Could not connect to database " + tag->getString("id"));
 143                         DelayReconnect();
 144                 }
 145         }
 146
 147         CullResult cull()
 148         {
 149                 this->SQLProvider::cull();
 150                 ServerInstance->Modules->DelService(*this);
 151                 return this->EventHandler::cull();
 152         }
 153
 154         ~SQLConn()
 155         {
 156                 SQLerror err(SQL_BAD_DBID);
 157                 if (qinprog)
 158                 {
 159                         qinprog->OnError(err);
 160                         delete qinprog;
 161                 }
 162                 for(std::deque<SQLQuery*>::iterator i = queue.begin(); i != queue.end(); i++)
 163                 {
 164                         SQLQuery* q = *i;
 165                         q->OnError(err);
 166                         delete q;
 167                 }
 168         }
 169
 170         virtual void HandleEvent(EventType et, int errornum)
 171         {
 172                 switch (et)
 173                 {
 174                         case EVENT_READ:
 175                         case EVENT_WRITE:
 176                                 DoEvent();
 177                         break;
 178
 179                         case EVENT_ERROR:
 180                                 DelayReconnect();
 181                 }
 182         }
 183
 184         std::string GetDSN()
 185         {
 186                 std::ostringstream conninfo("connect_timeout = '5'");
 187                 std::string item;
 188
 189                 if (conf->readString("host", item))
 190                         conninfo << " host = '" << item << "'";
 191
 192                 if (conf->readString("port", item))
 193                         conninfo << " port = '" << item << "'";
 194
 195                 if (conf->readString("name", item))
 196                         conninfo << " dbname = '" << item << "'";
 197
 198                 if (conf->readString("user", item))
 199                         conninfo << " user = '" << item << "'";
 200
 201                 if (conf->readString("pass", item))
 202                         conninfo << " password = '" << item << "'";
 203
 204                 if (conf->getBool("ssl"))
 205                         conninfo << " sslmode = 'require'";
 206                 else
 207                         conninfo << " sslmode = 'disable'";
 208
 209                 return conninfo.str();
 210         }
 211
 212         bool DoConnect()
 213         {
 214                 sql = PQconnectStart(GetDSN().c_str());
 215                 if (!sql)
 216                         return false;
 217
 218                 if(PQstatus(sql) == CONNECTION_BAD)
 219                         return false;
 220
 221                 if(PQsetnonblocking(sql, 1) == -1)
 222                         return false;
 223
 224                 /* OK, we've initalised the connection, now to get it hooked into the socket engine
 225                 * and then start polling it.
 226                 */
 227                 this->fd = PQsocket(sql);
 228
 229                 if(this->fd <= -1)
 230                         return false;
 231
 232                 if (!ServerInstance->SE->AddFd(this, FD_WANT_NO_WRITE | FD_WANT_NO_READ))
 233                 {
 234                         ServerInstance->Logs->Log("m_pgsql",DEBUG, "BUG: Couldn't add pgsql socket to socket engine");
 235                         return false;
 236                 }
 237
 238                 /* Socket all hooked into the engine, now to tell PgSQL to start connecting */
 239                 return DoPoll();
 240         }
 241
 242         bool DoPoll()
 243         {
 244                 switch(PQconnectPoll(sql))
 245                 {
 246                         case PGRES_POLLING_WRITING:
 247                                 ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_WRITE | FD_WANT_NO_READ);
 248                                 status = CWRITE;
 249                                 return true;
 250                         case PGRES_POLLING_READING:
 251                                 ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE);
 252                                 status = CREAD;
 253                                 return true;
 254                         case PGRES_POLLING_FAILED:
 255                                 return false;
 256                         case PGRES_POLLING_OK:
 257                                 ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE);
 258                                 status = WWRITE;
 259                                 DoConnectedPoll();
 260                         default:
 261                                 return true;
 262                 }
 263         }
 264
 265         void DoConnectedPoll()
 266         {
 267 restart:
 268                 while (!qinprog && !queue.empty())
 269                 {
 270                         /* There's no query currently in progress, and there's queries in the queue. */
 271                         DoQuery(queue.front());
 272                         queue.pop_front();
 273                 }
 274
 275                 if (PQconsumeInput(sql))
 276                 {
 277                         /* We just read stuff from the server, that counts as it being alive
 278                          * so update the idle-since time :p
 279                          */
 280                         idle = ServerInstance->Time();
 281
 282                         if (PQisBusy(sql))
 283                         {
 284                                 /* Nothing happens here */
 285                         }
 286                         else if (qinprog)
 287                         {
 288                                 /* Fetch the result.. */
 289                                 PGresult* result = PQgetResult(sql);
 290
 291                                 /* PgSQL would allow a query string to be sent which has multiple
 292                                  * queries in it, this isn't portable across database backends and
 293                                  * we don't want modules doing it. But just in case we make sure we
 294                                  * drain any results there are and just use the last one.
 295                                  * If the module devs are behaving there will only be one result.
 296                                  */
 297                                 while (PGresult* temp = PQgetResult(sql))
 298                                 {
 299                                         PQclear(result);
 300                                         result = temp;
 301                                 }
 302
 303                                 /* ..and the result */
 304                                 PgSQLresult reply(result);
 305                                 switch(PQresultStatus(result))
 306                                 {
 307                                         case PGRES_EMPTY_QUERY:
 308                                         case PGRES_BAD_RESPONSE:
 309                                         case PGRES_FATAL_ERROR:
 310                                         {
 311                                                 SQLerror err(SQL_QREPLY_FAIL, PQresultErrorMessage(result));
 312                                                 qinprog->OnError(err);
 313                                                 break;
 314                                         }
 315                                         default:
 316                                                 /* Other values are not errors */
 317                                                 qinprog->OnResult(reply);
 318                                 }
 319
 320                                 delete qinprog;
 321                                 qinprog = NULL;
 322                                 goto restart;
 323                         }
 324                 }
 325                 else
 326                 {
 327                         /* I think we'll assume this means the server died...it might not,
 328                          * but I think that any error serious enough we actually get here
 329                          * deserves to reconnect [/excuse]
 330                          * Returning true so the core doesn't try and close the connection.
 331                          */
 332                         DelayReconnect();
 333                 }
 334         }
 335
 336         bool DoResetPoll()
 337         {
 338                 switch(PQresetPoll(sql))
 339                 {
 340                         case PGRES_POLLING_WRITING:
 341                                 ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_WRITE | FD_WANT_NO_READ);
 342                                 status = CWRITE;
 343                                 return DoPoll();
 344                         case PGRES_POLLING_READING:
 345                                 ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE);
 346                                 status = CREAD;
 347                                 return true;
 348                         case PGRES_POLLING_FAILED:
 349                                 return false;
 350                         case PGRES_POLLING_OK:
 351                                 ServerInstance->SE->ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE);
 352                                 status = WWRITE;
 353                                 DoConnectedPoll();
 354                         default:
 355                                 return true;
 356                 }
 357         }
 358
 359         void DelayReconnect();
 360
 361         void DoEvent()
 362         {
 363                 if((status == CREAD) || (status == CWRITE))
 364                 {
 365                         DoPoll();
 366                 }
 367                 else if((status == RREAD) || (status == RWRITE))
 368                 {
 369                         DoResetPoll();
 370                 }
 371                 else
 372                 {
 373                         DoConnectedPoll();
 374                 }
 375         }
 376
 377         virtual std::string FormatQuery(const std::string& q, const ParamL& p)
 378         {
 379                 std::string res;
 380                 unsigned int param = 0;
 381                 for(std::string::size_type i = 0; i < q.length(); i++)
 382                 {
 383                         if (q[i] != '?')
 384                                 res.push_back(q[i]);
 385                         else
 386                         {
 387                                 // TODO numbered parameter support ('?1')
 388                                 if (param < p.size())
 389                                 {
 390                                         std::string parm = p[param++];
 391                                         char buffer[MAXBUF];
 392 #ifdef PGSQL_HAS_ESCAPECONN
 393                                         int error;
 394                                         PQescapeStringConn(sql, buffer, parm.c_str(), parm.length(), &error);
 395                                         if (error)
 396                                                 ServerInstance->Logs->Log("m_pgsql", DEBUG, "BUG: Apparently PQescapeStringConn() failed");
 397 #else
 398                                         PQescapeString         (buffer, parm.c_str(), parm.length());
 399 #endif
 400                                         res.append(buffer);
 401                                 }
 402                         }
 403                 }
 404                 return res;
 405         }
 406
 407         std::string FormatQuery(const std::string& q, const ParamM& p)
 408         {
 409                 std::string res;
 410                 for(std::string::size_type i = 0; i < q.length(); i++)
 411                 {
 412                         if (q[i] != '$')
 413                                 res.push_back(q[i]);
 414                         else
 415                         {
 416                                 std::string field;
 417                                 i++;
 418                                 while (i < q.length() && isalpha(q[i]))
 419                                         field.push_back(q[i++]);
 420                                 i--;
 421
 422                                 ParamM::const_iterator it = p.find(field);
 423                                 if (it != p.end())
 424                                 {
 425                                         std::string parm = it->second;
 426                                         char buffer[MAXBUF];
 427 #ifdef PGSQL_HAS_ESCAPECONN
 428                                         int error;
 429                                         PQescapeStringConn(sql, buffer, parm.c_str(), parm.length(), &error);
 430                                         if (error)
 431                                                 ServerInstance->Logs->Log("m_pgsql", DEBUG, "BUG: Apparently PQescapeStringConn() failed");
 432 #else
 433                                         PQescapeString         (buffer, parm.c_str(), parm.length());
 434 #endif
 435                                         res.append(buffer);
 436                                 }
 437                         }
 438                 }
 439                 return res;
 440         }
 441
 442         virtual void submit(SQLQuery *req)
 443         {
 444                 if (qinprog)
 445                 {
 446                         // wait your turn.
 447                         queue.push_back(req);
 448                 }
 449                 else
 450                 {
 451                         DoQuery(req);
 452                 }
 453         }
 454
 455         void DoQuery(SQLQuery* req)
 456         {
 457                 if (status != WREAD && status != WWRITE)
 458                 {
 459                         // whoops, not connected...
 460                         SQLerror err(SQL_BAD_CONN);
 461                         req->OnError(err);
 462                         delete req;
 463                         return;
 464                 }
 465
 466                 if(PQsendQuery(sql, req->query.c_str()))
 467                 {
 468                         qinprog = req;
 469                 }
 470                 else
 471                 {
 472                         SQLerror err(SQL_QSEND_FAIL, PQerrorMessage(sql));
 473                         req->OnError(err);
 474                         delete req;
 475                 }
 476         }
 477
 478         void Close()
 479         {
 480                 ServerInstance->SE->DelFd(this);
 481
 482                 if(sql)
 483                 {
 484                         PQfinish(sql);
 485                         sql = NULL;
 486                 }
 487         }
 488 };
 489
 490 class ModulePgSQL : public Module
 491 {
 492  public:
 493         ConnMap connections;
 494         ReconnectTimer* retimer;
 495
 496         ModulePgSQL()
 497         {
 498         }
 499
 500         void init()
 501         {
 502                 ReadConf();
 503
 504                 Implementation eventlist[] = { I_OnUnloadModule, I_OnRehash };
 505                 ServerInstance->Modules->Attach(eventlist, this, 2);
 506         }
 507
 508         virtual ~ModulePgSQL()
 509         {
 510                 if (retimer)
 511                         ServerInstance->Timers->DelTimer(retimer);
 512                 ClearAllConnections();
 513         }
 514
 515         virtual void OnRehash(User* user)
 516         {
 517                 ReadConf();
 518         }
 519
 520         void ReadConf()
 521         {
 522                 ConnMap conns;
 523                 ConfigTagList tags = ServerInstance->Config->ConfTags("database");
 524                 for(ConfigIter i = tags.first; i != tags.second; i++)
 525                 {
 526                         if (i->second->getString("module", "pgsql") != "pgsql")
 527                                 continue;
 528                         std::string id = i->second->getString("id");
 529                         ConnMap::iterator curr = connections.find(id);
 530                         if (curr == connections.end())
 531                         {
 532                                 SQLConn* conn = new SQLConn(this, i->second);
 533                                 conns.insert(std::make_pair(id, conn));
 534                                 ServerInstance->Modules->AddService(*conn);
 535                         }
 536                         else
 537                         {
 538                                 conns.insert(*curr);
 539                                 connections.erase(curr);
 540                         }
 541                 }
 542                 ClearAllConnections();
 543                 conns.swap(connections);
 544         }
 545
 546         void ClearAllConnections()
 547         {
 548                 for(ConnMap::iterator i = connections.begin(); i != connections.end(); i++)
 549                 {
 550                         i->second->cull();
 551                         delete i->second;
 552                 }
 553                 connections.clear();
 554         }
 555
 556         void OnUnloadModule(Module* mod)
 557         {
 558                 // TODO cancel queries that will have a bad vtable
 559         }
 560
 561         Version GetVersion()
 562         {
 563                 return Version("PostgreSQL Service Provider module for all other m_sql* modules, uses v2 of the SQL API", VF_VENDOR);
 564         }
 565 };
 566
 567 void ReconnectTimer::Tick(time_t time)
 568 {
 569         mod->retimer = NULL;
 570         mod->ReadConf();
 571 }
 572
 573 void SQLConn::DelayReconnect()
 574 {
 575         ModulePgSQL* mod = (ModulePgSQL*)(Module*)creator;
 576         ConnMap::iterator it = mod->connections.find(conf->getString("id"));
 577         if (it != mod->connections.end())
 578         {
 579                 mod->connections.erase(it);
 580                 ServerInstance->GlobalCulls.AddItem((EventHandler*)this);
 581                 if (!mod->retimer)
 582                 {
 583                         mod->retimer = new ReconnectTimer(mod);
 584                         ServerInstance->Timers->AddTimer(mod->retimer);
 585                 }
 586         }
 587 }
 588
 589 MODULE_INIT(ModulePgSQL)