src/modules/extra/m_pgsql.cpp

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