lib/WikiDB/backend/PearDB.php

   1 <?php
   2
   3 /*
   4  * Copyright 2002,2004,2005,2006 $ThePhpWikiProgrammingTeam
   5  *
   6  * This file is part of PhpWiki.
   7  *
   8  * PhpWiki is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License as published by
  10  * the Free Software Foundation; either version 2 of the License, or
  11  * (at your option) any later version.
  12  *
  13  * PhpWiki is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License along
  19  * with PhpWiki; if not, write to the Free Software Foundation, Inc.,
  20  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  21  */
  22
  23 require_once 'lib/WikiDB/backend.php';
  24
  25 class WikiDB_backend_PearDB
  26     extends WikiDB_backend
  27 {
  28     public $_dbh;
  29
  30     function __construct($dbparams)
  31     {
  32         // Find and include PEAR's DB.php. maybe we should force our private version again...
  33         // if DB would have exported its version number, it would be easier.
  34         @require_once('DB/common.php'); // Either our local pear copy or the system one
  35         $name = "escapeSimple";
  36         // TODO: apparently some Pear::Db version adds LIMIT 1,0 to getOne(),
  37         // which is invalid for "select version()"
  38         if (!in_array($name, get_class_methods("DB_common"))) {
  39             $finder = new FileFinder;
  40             $dir = dirname(__FILE__) . "/../../pear";
  41             $finder->_prepend_to_include_path($dir);
  42             include_once("$dir/DB/common.php"); // use our version instead.
  43             if (!in_array($name, get_class_methods("DB_common"))) {
  44                 $pearFinder = new PearFileFinder("lib/pear");
  45                 $pearFinder->includeOnce('DB.php');
  46             } else {
  47                 include_once("$dir/DB.php");
  48             }
  49         } else {
  50             include_once 'DB.php';
  51         }
  52
  53         // Install filter to handle bogus error notices from buggy DB.php's.
  54         // TODO: check the Pear_DB version, but how?
  55         if (DEBUG) {
  56             global $ErrorManager;
  57             $ErrorManager->pushErrorHandler(new WikiMethodCb($this, '_pear_notice_filter'));
  58             $this->_pearerrhandler = true;
  59         }
  60
  61         // Open connection to database
  62         $this->_dsn = $dbparams['dsn'];
  63         $this->_dbparams = $dbparams;
  64         $this->_lock_count = 0;
  65
  66         // persistent is usually a DSN option: we override it with a config value.
  67         //   phptype://username:password@hostspec/database?persistent=false
  68         $dboptions = array('persistent' => DATABASE_PERSISTENT,
  69             'debug' => 2);
  70         //if (preg_match('/^pgsql/', $this->_dsn)) $dboptions['persistent'] = false;
  71         $this->_dbh = DB::connect($this->_dsn, $dboptions);
  72         $dbh = &$this->_dbh;
  73         if (DB::isError($dbh)) {
  74             trigger_error(sprintf("Can't connect to database: %s",
  75                     $this->_pear_error_message($dbh)),
  76                 isset($dbparams['_tryroot_from_upgrade']) // hack!
  77                     ? E_USER_WARNING : E_USER_ERROR);
  78             if (isset($dbparams['_tryroot_from_upgrade']))
  79                 return;
  80         }
  81         $dbh->setErrorHandling(PEAR_ERROR_CALLBACK,
  82             array($this, '_pear_error_callback'));
  83         $dbh->setFetchMode(DB_FETCHMODE_ASSOC);
  84
  85         $prefix = isset($dbparams['prefix']) ? $dbparams['prefix'] : '';
  86         $this->_table_names
  87             = array('page_tbl' => $prefix . 'page',
  88             'version_tbl' => $prefix . 'version',
  89             'link_tbl' => $prefix . 'link',
  90             'recent_tbl' => $prefix . 'recent',
  91             'nonempty_tbl' => $prefix . 'nonempty');
  92         $page_tbl = $this->_table_names['page_tbl'];
  93         $version_tbl = $this->_table_names['version_tbl'];
  94         $this->page_tbl_fields = "$page_tbl.id AS id, $page_tbl.pagename AS pagename, $page_tbl.hits AS hits";
  95         $this->version_tbl_fields = "$version_tbl.version AS version, $version_tbl.mtime AS mtime, " .
  96             "$version_tbl.minor_edit AS minor_edit, $version_tbl.content AS content, $version_tbl.versiondata AS versiondata";
  97
  98         $this->_expressions
  99             = array('maxmajor' => "MAX(CASE WHEN minor_edit=0 THEN version END)",
 100             'maxminor' => "MAX(CASE WHEN minor_edit<>0 THEN version END)",
 101             'maxversion' => "MAX(version)",
 102             'notempty' => "<>''",
 103             'iscontent' => "content<>''");
 104
 105     }
 106
 107     /**
 108      * Close database connection.
 109      */
 110     function close()
 111     {
 112         if (!$this->_dbh)
 113             return;
 114         if ($this->_lock_count) {
 115             trigger_error("WARNING: database still locked " .
 116                     '(lock_count = $this->_lock_count)' . "\n<br />",
 117                 E_USER_WARNING);
 118         }
 119         $this->_dbh->setErrorHandling(PEAR_ERROR_PRINT); // prevent recursive loops.
 120         $this->unlock('force');
 121
 122         $this->_dbh->disconnect();
 123
 124         if (!empty($this->_pearerrhandler)) {
 125             $GLOBALS['ErrorManager']->popErrorHandler();
 126         }
 127     }
 128
 129     /*
 130      * Fast test for wikipage.
 131      */
 132     function is_wiki_page($pagename)
 133     {
 134         $dbh = &$this->_dbh;
 135         extract($this->_table_names);
 136         return $dbh->getOne(sprintf("SELECT $page_tbl.id AS id"
 137                 . " FROM $nonempty_tbl, $page_tbl"
 138                 . " WHERE $nonempty_tbl.id=$page_tbl.id"
 139                 . "   AND pagename='%s'",
 140             $dbh->escapeSimple($pagename)));
 141     }
 142
 143     function get_all_pagenames()
 144     {
 145         $dbh = &$this->_dbh;
 146         extract($this->_table_names);
 147         return $dbh->getCol("SELECT pagename"
 148             . " FROM $nonempty_tbl, $page_tbl"
 149             . " WHERE $nonempty_tbl.id=$page_tbl.id");
 150     }
 151
 152     /*
 153      * filter (nonempty pages) currently ignored
 154      */
 155     function numPages($filter = false, $exclude = '')
 156     {
 157         $dbh = &$this->_dbh;
 158         extract($this->_table_names);
 159         return $dbh->getOne("SELECT count(*)"
 160             . " FROM $nonempty_tbl, $page_tbl"
 161             . " WHERE $nonempty_tbl.id=$page_tbl.id");
 162     }
 163
 164     function increaseHitCount($pagename)
 165     {
 166         $dbh = &$this->_dbh;
 167         // Hits is the only thing we can update in a fast manner.
 168         // Note that this will fail silently if the page does not
 169         // have a record in the page table.  Since it's just the
 170         // hit count, who cares?
 171         $dbh->query(sprintf("UPDATE %s SET hits=hits+1 WHERE pagename='%s'",
 172             $this->_table_names['page_tbl'],
 173             $dbh->escapeSimple($pagename)));
 174     }
 175
 176     /*
 177      * Read page information from database.
 178      */
 179     function get_pagedata($pagename)
 180     {
 181         $dbh = &$this->_dbh;
 182         //trigger_error("GET_PAGEDATA $pagename", E_USER_NOTICE);
 183         $result = $dbh->getRow(sprintf("SELECT hits,pagedata FROM %s WHERE pagename='%s'",
 184                 $this->_table_names['page_tbl'],
 185                 $dbh->escapeSimple($pagename)),
 186             DB_FETCHMODE_ASSOC);
 187         return $result ? $this->_extract_page_data($result) : false;
 188     }
 189
 190     public function  _extract_page_data($data)
 191     {
 192         if (empty($data)) return array();
 193         elseif (empty($data['pagedata'])) return $data; else {
 194             $data = array_merge($data, $this->_unserialize($data['pagedata']));
 195             unset($data['pagedata']);
 196             return $data;
 197         }
 198     }
 199
 200     function update_pagedata($pagename, $newdata)
 201     {
 202         $dbh = &$this->_dbh;
 203         $page_tbl = $this->_table_names['page_tbl'];
 204
 205         // Hits is the only thing we can update in a fast manner.
 206         if (count($newdata) == 1 && isset($newdata['hits'])) {
 207             // Note that this will fail silently if the page does not
 208             // have a record in the page table.  Since it's just the
 209             // hit count, who cares?
 210             $dbh->query(sprintf("UPDATE $page_tbl SET hits=%d WHERE pagename='%s'",
 211                 $newdata['hits'], $dbh->escapeSimple($pagename)));
 212             return;
 213         }
 214
 215         $this->lock(array($page_tbl), true);
 216         $data = $this->get_pagedata($pagename);
 217         if (!$data) {
 218             $data = array();
 219             $this->_get_pageid($pagename, true); // Creates page record
 220         }
 221
 222         $hits = (empty($data['hits'])) ? 0 : (int)$data['hits'];
 223         unset($data['hits']);
 224
 225         foreach ($newdata as $key => $val) {
 226             if ($key == 'hits')
 227                 $hits = (int)$val;
 228             else if (empty($val))
 229                 unset($data[$key]);
 230             else
 231                 $data[$key] = $val;
 232         }
 233
 234         /* Portability issue -- not all DBMS supports huge strings
 235          * so we need to 'bind' instead of building a simple SQL statment.
 236          * Note that we do not need to escapeSimple when we bind
 237         $dbh->query(sprintf("UPDATE $page_tbl"
 238                             . " SET hits=%d, pagedata='%s'"
 239                             . " WHERE pagename='%s'",
 240                             $hits,
 241                             $dbh->escapeSimple($this->_serialize($data)),
 242                             $dbh->escapeSimple($pagename)));
 243         */
 244         $dbh->query("UPDATE $page_tbl"
 245                 . " SET hits=?, pagedata=?"
 246                 . " WHERE pagename=?",
 247             array($hits, $this->_serialize($data), $pagename));
 248         $this->unlock(array($page_tbl));
 249     }
 250
 251     function get_cached_html($pagename)
 252     {
 253         $dbh = &$this->_dbh;
 254         $page_tbl = $this->_table_names['page_tbl'];
 255         return $dbh->GetOne(sprintf("SELECT cached_html FROM $page_tbl WHERE pagename='%s'",
 256             $dbh->escapeSimple($pagename)));
 257     }
 258
 259     function set_cached_html($pagename, $data)
 260     {
 261         $dbh = &$this->_dbh;
 262         $page_tbl = $this->_table_names['page_tbl'];
 263         $dbh->query("UPDATE $page_tbl"
 264                 . " SET cached_html=?"
 265                 . " WHERE pagename=?",
 266             array($data, $pagename));
 267     }
 268
 269     function _get_pageid($pagename, $create_if_missing = false)
 270     {
 271         // check id_cache
 272         global $request;
 273         $cache =& $request->_dbi->_cache->_id_cache;
 274         if (isset($cache[$pagename])) {
 275             if ($cache[$pagename] or !$create_if_missing) {
 276                 return $cache[$pagename];
 277             }
 278         }
 279
 280         // attributes play this game.
 281         if ($pagename === '') return 0;
 282
 283         $dbh = &$this->_dbh;
 284         $page_tbl = $this->_table_names['page_tbl'];
 285
 286         $query = sprintf("SELECT id FROM $page_tbl WHERE pagename='%s'",
 287             $dbh->escapeSimple($pagename));
 288
 289         if (!$create_if_missing)
 290             return $dbh->getOne($query);
 291
 292         $id = $dbh->getOne($query);
 293         if (empty($id)) {
 294             $this->lock(array($page_tbl), true); // write lock
 295             $max_id = $dbh->getOne("SELECT MAX(id) FROM $page_tbl");
 296             $id = $max_id + 1;
 297             // requires createSequence and on mysql lock the interim table ->getSequenceName
 298             //$id = $dbh->nextId($page_tbl . "_id");
 299             $dbh->query(sprintf("INSERT INTO $page_tbl"
 300                     . " (id,pagename,hits)"
 301                     . " VALUES (%d,'%s',0)",
 302                 $id, $dbh->escapeSimple($pagename)));
 303             $this->unlock(array($page_tbl));
 304         }
 305         return $id;
 306     }
 307
 308     function get_latest_version($pagename)
 309     {
 310         $dbh = &$this->_dbh;
 311         extract($this->_table_names);
 312         return
 313             (int)$dbh->getOne(sprintf("SELECT latestversion"
 314                     . " FROM $page_tbl, $recent_tbl"
 315                     . " WHERE $page_tbl.id=$recent_tbl.id"
 316                     . "  AND pagename='%s'",
 317                 $dbh->escapeSimple($pagename)));
 318     }
 319
 320     function get_previous_version($pagename, $version)
 321     {
 322         $dbh = &$this->_dbh;
 323         extract($this->_table_names);
 324
 325         return
 326             (int)$dbh->getOne(sprintf("SELECT version"
 327                     . " FROM $version_tbl, $page_tbl"
 328                     . " WHERE $version_tbl.id=$page_tbl.id"
 329                     . "  AND pagename='%s'"
 330                     . "  AND version < %d"
 331                     . " ORDER BY version DESC",
 332                 /* Non portable and useless anyway with getOne
 333                 . " LIMIT 1",
 334                 */
 335                 $dbh->escapeSimple($pagename),
 336                 $version));
 337     }
 338
 339     /**
 340      * Get version data.
 341      *
 342      * @param string $pagename Name of the page
 343      * @param int $version Which version to get
 344      * @param bool $want_content Do we need content?
 345      *
 346      * @return array hash The version data, or false if specified version does not
 347      *              exist.
 348      */
 349     function get_versiondata($pagename, $version, $want_content = false)
 350     {
 351         $dbh = &$this->_dbh;
 352         extract($this->_table_names);
 353         extract($this->_expressions);
 354
 355         assert(is_string($pagename) and $pagename != '');
 356         assert($version > 0);
 357
 358         // FIXME: optimization: sometimes don't get page data?
 359         if ($want_content) {
 360             $fields = $this->page_tbl_fields
 361                 . ",$page_tbl.pagedata AS pagedata,"
 362                 . $this->version_tbl_fields;
 363         } else {
 364             $fields = $this->page_tbl_fields . ","
 365                 . "mtime, minor_edit, versiondata,"
 366                 . "$iscontent AS have_content";
 367         }
 368
 369         $result = $dbh->getRow(sprintf("SELECT $fields"
 370                     . " FROM $page_tbl, $version_tbl"
 371                     . " WHERE $page_tbl.id=$version_tbl.id"
 372                     . "  AND pagename='%s'"
 373                     . "  AND version=%d",
 374                 $dbh->escapeSimple($pagename), $version),
 375             DB_FETCHMODE_ASSOC);
 376
 377         return $this->_extract_version_data($result);
 378     }
 379
 380     function _extract_version_data($query_result)
 381     {
 382         if (!$query_result)
 383             return false;
 384
 385         /* Earlier versions (<= 1.3.7) stored the version data in base64.
 386            This could be done here or in upgrade.
 387         */
 388         if (!strstr($query_result['versiondata'], ":")) {
 389             $query_result['versiondata'] =
 390                 base64_decode($query_result['versiondata']);
 391         }
 392         $data = $this->_unserialize($query_result['versiondata']);
 393
 394         $data['mtime'] = $query_result['mtime'];
 395         $data['is_minor_edit'] = !empty($query_result['minor_edit']);
 396
 397         if (isset($query_result['content']))
 398             $data['%content'] = $query_result['content'];
 399         elseif ($query_result['have_content'])
 400             $data['%content'] = true; else
 401             $data['%content'] = '';
 402
 403         // FIXME: this is ugly.
 404         if (isset($query_result['pagedata'])) {
 405             // Query also includes page data.
 406             // We might as well send that back too...
 407             unset($query_result['versiondata']);
 408             $data['%pagedata'] = $this->_extract_page_data($query_result);
 409         }
 410         return $data;
 411     }
 412
 413     /*
 414      * Create a new revision of a page.
 415      */
 416     function set_versiondata($pagename, $version, $data)
 417     {
 418         $dbh = &$this->_dbh;
 419         $version_tbl = $this->_table_names['version_tbl'];
 420
 421         $minor_edit = (int)!empty($data['is_minor_edit']);
 422         unset($data['is_minor_edit']);
 423
 424         $mtime = (int)$data['mtime'];
 425         unset($data['mtime']);
 426         assert(!empty($mtime));
 427
 428         $content = isset($data['%content']) ? (string)$data['%content'] : '';
 429         unset($data['%content']);
 430         unset($data['%pagedata']);
 431
 432         $this->lock();
 433         $id = $this->_get_pageid($pagename, true);
 434
 435         $dbh->query(sprintf("DELETE FROM $version_tbl"
 436                 . " WHERE id=%d AND version=%d",
 437             $id, $version));
 438         // generic slow PearDB bind eh quoting.
 439         $dbh->query("INSERT INTO $version_tbl"
 440                 . " (id,version,mtime,minor_edit,content,versiondata)"
 441                 . " VALUES(?, ?, ?, ?, ?, ?)",
 442             array($id, $version, $mtime, $minor_edit, $content,
 443                 $this->_serialize($data)));
 444
 445         $this->_update_recent_table($id);
 446         $this->_update_nonempty_table($id);
 447
 448         $this->unlock();
 449     }
 450
 451     /*
 452      * Delete an old revision of a page.
 453      */
 454     function delete_versiondata($pagename, $version)
 455     {
 456         $dbh = &$this->_dbh;
 457         extract($this->_table_names);
 458
 459         $this->lock();
 460         if (($id = $this->_get_pageid($pagename))) {
 461             $dbh->query("DELETE FROM $version_tbl"
 462                 . " WHERE id=$id AND version=$version");
 463             $this->_update_recent_table($id);
 464             // This shouldn't be needed (as long as the latestversion
 465             // never gets deleted.)  But, let's be safe.
 466             $this->_update_nonempty_table($id);
 467         }
 468         $this->unlock();
 469     }
 470
 471     /*
 472      * Delete page completely from the database.
 473      */
 474     function purge_page($pagename)
 475     {
 476         $dbh = &$this->_dbh;
 477         extract($this->_table_names);
 478
 479         $this->lock();
 480         if (($id = $this->_get_pageid($pagename, false))) {
 481             $dbh->query("DELETE FROM $nonempty_tbl WHERE id=$id");
 482             $dbh->query("DELETE FROM $recent_tbl   WHERE id=$id");
 483             $dbh->query("DELETE FROM $version_tbl  WHERE id=$id");
 484             $dbh->query("DELETE FROM $link_tbl     WHERE linkfrom=$id");
 485             $nlinks = $dbh->getOne("SELECT COUNT(*) FROM $link_tbl WHERE linkto=$id");
 486             if ($nlinks) {
 487                 // We're still in the link table (dangling link) so we can't delete this
 488                 // altogether.
 489                 $dbh->query("UPDATE $page_tbl SET hits=0, pagedata='' WHERE id=$id");
 490                 $result = 0;
 491             } else {
 492                 $dbh->query("DELETE FROM $page_tbl WHERE id=$id");
 493                 $result = 1;
 494             }
 495             $this->_update_recent_table();
 496             $this->_update_nonempty_table();
 497         } else {
 498             $result = -1; // already purged or not existing
 499         }
 500         $this->unlock();
 501         return $result;
 502     }
 503
 504     /*
 505      * Update link table.
 506      * on DEBUG: delete old, deleted links from page
 507      */
 508     function set_links($pagename, $links)
 509     {
 510         // Update link table.
 511         // FIXME: optimize: mysql can do this all in one big INSERT/REPLACE.
 512
 513         $dbh = &$this->_dbh;
 514         extract($this->_table_names);
 515
 516         $this->lock();
 517         $pageid = $this->_get_pageid($pagename, true);
 518
 519         $dbh->query("DELETE FROM $link_tbl WHERE linkfrom=$pageid");
 520         if ($links) {
 521             $linkseen = array();
 522             foreach ($links as $link) {
 523                 $linkto = $link['linkto'];
 524                 if ($linkto === "") { // ignore attributes
 525                     continue;
 526                 }
 527                 if (isset($link['relation']))
 528                     $relation = $this->_get_pageid($link['relation'], true);
 529                 else
 530                     $relation = 0;
 531                 // avoid duplicates
 532                 if (isset($linkseen[$linkto]) and !$relation) {
 533                     continue;
 534                 }
 535                 if (!$relation) {
 536                     $linkseen[$linkto] = true;
 537                 }
 538                 $linkid = $this->_get_pageid($linkto, true);
 539                 if (!$linkid) {
 540                     echo("No link for $linkto on page $pagename");
 541                     //printSimpleTrace(debug_backtrace());
 542                     trigger_error("No link for $linkto on page $pagename");
 543                 }
 544                 assert($linkid);
 545                 $dbh->query("INSERT INTO $link_tbl (linkfrom, linkto, relation)"
 546                     . " VALUES ($pageid, $linkid, $relation)");
 547             }
 548             unset($linkseen);
 549         }
 550         $this->unlock();
 551     }
 552
 553     /*
 554      * Find pages which link to or are linked from a page.
 555      *
 556      * TESTME relations: get_links is responsible to add the relation to the pagehash
 557      * as 'linkrelation' key as pagename. See WikiDB_PageIterator::next
 558      *   if (isset($next['linkrelation']))
 559      */
 560     function get_links($pagename, $reversed = true, $include_empty = false,
 561                        $sortby = '', $limit = '', $exclude = '',
 562                        $want_relations = false)
 563     {
 564         $dbh = &$this->_dbh;
 565         extract($this->_table_names);
 566
 567         if ($reversed)
 568             list($have, $want) = array('linkee', 'linker');
 569         else
 570             list($have, $want) = array('linker', 'linkee');
 571         $orderby = $this->sortby($sortby, 'db', array('pagename'));
 572         if ($orderby) $orderby = " ORDER BY $want." . $orderby;
 573         if ($exclude) // array of pagenames
 574             $exclude = " AND $want.pagename NOT IN " . $this->_sql_set($exclude);
 575         else
 576             $exclude = '';
 577
 578         $qpagename = $dbh->escapeSimple($pagename);
 579         $sql = "SELECT $want.id AS id, $want.pagename AS pagename, "
 580             . ($want_relations ? " related.pagename as linkrelation" : " $want.hits AS hits")
 581             . " FROM "
 582             . (!$include_empty ? "$nonempty_tbl, " : '')
 583             . " $page_tbl linkee, $page_tbl linker, $link_tbl "
 584             . ($want_relations ? " JOIN $page_tbl related ON ($link_tbl.relation=related.id)" : '')
 585             . " WHERE linkfrom=linker.id AND linkto=linkee.id"
 586             . " AND $have.pagename='$qpagename'"
 587             . (!$include_empty ? " AND $nonempty_tbl.id=$want.id" : "")
 588             //. " GROUP BY $want.id"
 589             . $exclude
 590             . $orderby;
 591         if ($limit) {
 592             // extract from,count from limit
 593             list($from, $count) = $this->limit($limit);
 594             $result = $dbh->limitQuery($sql, $from, $count);
 595         } else {
 596             $result = $dbh->query($sql);
 597         }
 598
 599         return new WikiDB_backend_PearDB_iter($this, $result);
 600     }
 601
 602     /*
 603      * Find if a page links to another page
 604      */
 605     function exists_link($pagename, $link, $reversed = false)
 606     {
 607         $dbh = &$this->_dbh;
 608         extract($this->_table_names);
 609
 610         if ($reversed)
 611             list($have, $want) = array('linkee', 'linker');
 612         else
 613             list($have, $want) = array('linker', 'linkee');
 614         $qpagename = $dbh->escapeSimple($pagename);
 615         $qlink = $dbh->escapeSimple($link);
 616         $row = $dbh->GetRow("SELECT CASE WHEN $want.pagename='$qlink' THEN 1 ELSE 0 END as result"
 617             . " FROM $link_tbl, $page_tbl linker, $page_tbl linkee, $nonempty_tbl"
 618             . " WHERE linkfrom=linker.id AND linkto=linkee.id"
 619             . " AND $have.pagename='$qpagename'"
 620             . " AND $want.pagename='$qlink'");
 621         return $row['result'];
 622     }
 623
 624     public function get_all_pages($include_empty = false,
 625                                   $sortby = '', $limit = '', $exclude = '')
 626     {
 627         $dbh = &$this->_dbh;
 628         extract($this->_table_names);
 629         $orderby = $this->sortby($sortby, 'db');
 630         if ($orderby) $orderby = ' ORDER BY ' . $orderby;
 631         if ($exclude) { // array of pagenames
 632             $exclude = " AND $page_tbl.pagename NOT IN " . $this->_sql_set($exclude);
 633         } else {
 634             $exclude = '';
 635         }
 636
 637         if (strstr($orderby, 'mtime ')) { // multiple columns possible
 638             if ($include_empty) {
 639                 $sql = "SELECT "
 640                     . $this->page_tbl_fields
 641                     . " FROM $page_tbl, $recent_tbl, $version_tbl"
 642                     . " WHERE $page_tbl.id=$recent_tbl.id"
 643                     . " AND $page_tbl.id=$version_tbl.id AND latestversion=version"
 644                     . $exclude
 645                     . $orderby;
 646             } else {
 647                 $sql = "SELECT "
 648                     . $this->page_tbl_fields
 649                     . " FROM $nonempty_tbl, $page_tbl, $recent_tbl, $version_tbl"
 650                     . " WHERE $nonempty_tbl.id=$page_tbl.id"
 651                     . " AND $page_tbl.id=$recent_tbl.id"
 652                     . " AND $page_tbl.id=$version_tbl.id AND latestversion=version"
 653                     . $exclude
 654                     . $orderby;
 655             }
 656         } else {
 657             if ($include_empty) {
 658                 $sql = "SELECT "
 659                     . $this->page_tbl_fields
 660                     . " FROM $page_tbl"
 661                     . ($exclude ? " WHERE $exclude" : '')
 662                     . $orderby;
 663             } else {
 664                 $sql = "SELECT "
 665                     . $this->page_tbl_fields
 666                     . " FROM $nonempty_tbl, $page_tbl"
 667                     . " WHERE $nonempty_tbl.id=$page_tbl.id"
 668                     . $exclude
 669                     . $orderby;
 670             }
 671         }
 672         if ($limit && $orderby) {
 673             // extract from,count from limit
 674             list($from, $count) = $this->limit($limit);
 675             $result = $dbh->limitQuery($sql, $from, $count);
 676             $options = array('limit_by_db' => 1);
 677         } else {
 678             $result = $dbh->query($sql);
 679             $options = array('limit_by_db' => 0);
 680         }
 681         return new WikiDB_backend_PearDB_iter($this, $result, $options);
 682     }
 683
 684     /*
 685      * Title and fulltext search.
 686      */
 687     public function text_search($search, $fulltext = false,
 688                                 $sortby = '', $limit = '', $exclude = '')
 689     {
 690         $dbh = &$this->_dbh;
 691         extract($this->_table_names);
 692         $orderby = $this->sortby($sortby, 'db');
 693         if ($orderby) $orderby = ' ORDER BY ' . $orderby;
 694         //else " ORDER BY rank($field, to_tsquery('$searchon')) DESC";
 695
 696         $searchclass = get_class($this) . "_search";
 697         // no need to define it everywhere and then fallback. memory!
 698         if (!class_exists($searchclass))
 699             $searchclass = "WikiDB_backend_PearDB_search";
 700         $searchobj = new $searchclass($search, $dbh);
 701
 702         $table = "$nonempty_tbl, $page_tbl";
 703         $join_clause = "$nonempty_tbl.id=$page_tbl.id";
 704         $fields = $this->page_tbl_fields;
 705
 706         if ($fulltext) {
 707             $table .= ", $recent_tbl";
 708             $join_clause .= " AND $page_tbl.id=$recent_tbl.id";
 709
 710             $table .= ", $version_tbl";
 711             $join_clause .= " AND $page_tbl.id=$version_tbl.id AND latestversion=version";
 712
 713             $fields .= ", $page_tbl.pagedata as pagedata, " . $this->version_tbl_fields;
 714             $callback = new WikiMethodCb($searchobj, "_fulltext_match_clause");
 715         } else {
 716             $callback = new WikiMethodCb($searchobj, "_pagename_match_clause");
 717         }
 718         $search_clause = $search->makeSqlClauseObj($callback);
 719         $sql = "SELECT $fields FROM $table"
 720             . " WHERE $join_clause"
 721             . "  AND ($search_clause)"
 722             . $orderby;
 723         if ($limit) {
 724             list($from, $count) = $this->limit($limit);
 725             $result = $dbh->limitQuery($sql, $from, $count);
 726         } else {
 727             $result = $dbh->query($sql);
 728         }
 729
 730         $iter = new WikiDB_backend_PearDB_iter($this, $result);
 731         $iter->stoplisted = @$searchobj->stoplisted;
 732         return $iter;
 733     }
 734
 735     //Todo: check if the better Mysql MATCH operator is supported,
 736     // (ranked search) and also google like expressions.
 737     function _sql_match_clause($word)
 738     {
 739         $word = preg_replace('/(?=[%_\\\\])/', "\\", $word);
 740         $word = $this->_dbh->escapeSimple($word);
 741         //$page_tbl = $this->_table_names['page_tbl'];
 742         //Note: Mysql 4.1.0 has a bug which fails with binary fields.
 743         //      e.g. if word is lowercased.
 744         // http://bugs.mysql.com/bug.php?id=1491
 745         return "LOWER(pagename) LIKE '%$word%'";
 746     }
 747
 748     function _sql_casematch_clause($word)
 749     {
 750         $word = preg_replace('/(?=[%_\\\\])/', "\\", $word);
 751         $word = $this->_dbh->escapeSimple($word);
 752         return "pagename LIKE '%$word%'";
 753     }
 754
 755     function _fullsearch_sql_match_clause($word)
 756     {
 757         $word = preg_replace('/(?=[%_\\\\])/', "\\", $word);
 758         $word = $this->_dbh->escapeSimple($word);
 759         //$page_tbl = $this->_table_names['page_tbl'];
 760         //Mysql 4.1.1 has a bug which fails here if word is lowercased.
 761         return "LOWER(pagename) LIKE '%$word%' OR content LIKE '%$word%'";
 762     }
 763
 764     function _fullsearch_sql_casematch_clause($word)
 765     {
 766         $word = preg_replace('/(?=[%_\\\\])/', "\\", $word);
 767         $word = $this->_dbh->escapeSimple($word);
 768         return "pagename LIKE '%$word%' OR content LIKE '%$word%'";
 769     }
 770
 771     /*
 772      * Find highest or lowest hit counts.
 773      */
 774     public function most_popular($limit = 20, $sortby = '-hits')
 775     {
 776         $dbh = &$this->_dbh;
 777         extract($this->_table_names);
 778         if ($limit < 0) {
 779             $order = "hits ASC";
 780             $limit = -$limit;
 781             $where = "";
 782         } else {
 783             $order = "hits DESC";
 784             $where = " AND hits > 0";
 785         }
 786         $orderby = '';
 787         if ($sortby != '-hits') {
 788             if ($order = $this->sortby($sortby, 'db'))
 789                 $orderby = " ORDER BY " . $order;
 790         } else {
 791             $orderby = " ORDER BY $order";
 792         }
 793         $sql = "SELECT "
 794             . $this->page_tbl_fields
 795             . " FROM $nonempty_tbl, $page_tbl"
 796             . " WHERE $nonempty_tbl.id=$page_tbl.id"
 797             . $where
 798             . $orderby;
 799         if ($limit) {
 800             list($from, $count) = $this->limit($limit);
 801             $result = $dbh->limitQuery($sql, $from, $count);
 802         } else {
 803             $result = $dbh->query($sql);
 804         }
 805         return new WikiDB_backend_PearDB_iter($this, $result);
 806     }
 807
 808     /*
 809      * Find recent changes.
 810      */
 811     public function most_recent($params)
 812     {
 813         $limit = 0;
 814         $since = 0;
 815         $include_minor_revisions = false;
 816         $exclude_major_revisions = false;
 817         $include_all_revisions = false;
 818         extract($params);
 819
 820         $dbh = &$this->_dbh;
 821         extract($this->_table_names);
 822
 823         $pick = array();
 824         if ($since)
 825             $pick[] = "mtime >= $since";
 826
 827         if ($include_all_revisions) {
 828             // Include all revisions of each page.
 829             $table = "$page_tbl, $version_tbl";
 830             $join_clause = "$page_tbl.id=$version_tbl.id";
 831
 832             if ($exclude_major_revisions) {
 833                 // Include only minor revisions
 834                 $pick[] = "minor_edit <> 0";
 835             } elseif (!$include_minor_revisions) {
 836                 // Include only major revisions
 837                 $pick[] = "minor_edit = 0";
 838             }
 839         } else {
 840             $table = "$page_tbl, $recent_tbl";
 841             $join_clause = "$page_tbl.id=$recent_tbl.id";
 842             $table .= ", $version_tbl";
 843             $join_clause .= " AND $version_tbl.id=$page_tbl.id";
 844
 845             if ($exclude_major_revisions) {
 846                 // Include only most recent minor revision
 847                 $pick[] = 'version=latestminor';
 848             } elseif (!$include_minor_revisions) {
 849                 // Include only most recent major revision
 850                 $pick[] = 'version=latestmajor';
 851             } else {
 852                 // Include only the latest revision (whether major or minor).
 853                 $pick[] = 'version=latestversion';
 854             }
 855         }
 856         $order = "DESC";
 857         if ($limit < 0) {
 858             $order = "ASC";
 859             $limit = -$limit;
 860         }
 861         $where_clause = $join_clause;
 862         if ($pick)
 863             $where_clause .= " AND " . join(" AND ", $pick);
 864         $sql = "SELECT "
 865             . $this->page_tbl_fields . ", " . $this->version_tbl_fields
 866             . " FROM $table"
 867             . " WHERE $where_clause"
 868             . " ORDER BY mtime $order";
 869         // FIXME: use SQL_BUFFER_RESULT for mysql?
 870         if ($limit) {
 871             list($from, $count) = $this->limit($limit);
 872             $result = $dbh->limitQuery($sql, $from, $count);
 873         } else {
 874             $result = $dbh->query($sql);
 875         }
 876         return new WikiDB_backend_PearDB_iter($this, $result);
 877     }
 878
 879     /*
 880      * Find referenced empty pages.
 881      */
 882     function wanted_pages($exclude_from = '', $exclude = '', $sortby = '', $limit = '')
 883     {
 884         $dbh = &$this->_dbh;
 885         extract($this->_table_names);
 886         if ($orderby = $this->sortby($sortby, 'db', array('pagename', 'wantedfrom')))
 887             $orderby = 'ORDER BY ' . $orderby;
 888
 889         if ($exclude_from) // array of pagenames
 890             $exclude_from = " AND pp.pagename NOT IN " . $this->_sql_set($exclude_from);
 891         if ($exclude) // array of pagenames
 892             $exclude = " AND p.pagename NOT IN " . $this->_sql_set($exclude);
 893
 894         $sql = "SELECT p.pagename, pp.pagename AS wantedfrom"
 895             . " FROM $page_tbl p, $link_tbl linked"
 896             . " LEFT JOIN $page_tbl pp ON linked.linkto = pp.id"
 897             . " LEFT JOIN $nonempty_tbl ne ON linked.linkto = ne.id"
 898             . " WHERE ne.id IS NULL"
 899             . " AND p.id = linked.linkfrom"
 900             . $exclude_from
 901             . $exclude
 902             . $orderby;
 903         if ($limit) {
 904             list($from, $count) = $this->limit($limit);
 905             $result = $dbh->limitQuery($sql, $from, $count * 3);
 906         } else {
 907             $result = $dbh->query($sql);
 908         }
 909         return new WikiDB_backend_PearDB_generic_iter($this, $result);
 910     }
 911
 912     function _sql_set(&$pagenames)
 913     {
 914         $s = '(';
 915         foreach ($pagenames as $p) {
 916             $s .= ("'" . $this->_dbh->escapeSimple($p) . "',");
 917         }
 918         return substr($s, 0, -1) . ")";
 919     }
 920
 921     /*
 922      * Rename page in the database.
 923      */
 924     function rename_page($pagename, $to)
 925     {
 926         $dbh = &$this->_dbh;
 927         extract($this->_table_names);
 928
 929         $this->lock();
 930         if (($id = $this->_get_pageid($pagename, false))) {
 931             if ($new = $this->_get_pageid($to, false)) {
 932                 // Cludge Alert!
 933                 // This page does not exist (already verified before), but exists in the page table.
 934                 // So we delete this page.
 935                 $dbh->query("DELETE FROM $nonempty_tbl WHERE id=$new");
 936                 $dbh->query("DELETE FROM $recent_tbl WHERE id=$new");
 937                 $dbh->query("DELETE FROM $version_tbl WHERE id=$new");
 938                 // We have to fix all referring tables to the old id
 939                 $dbh->query("UPDATE $link_tbl SET linkfrom=$id WHERE linkfrom=$new");
 940                 $dbh->query("UPDATE $link_tbl SET linkto=$id WHERE linkto=$new");
 941                 $dbh->query("DELETE FROM $page_tbl WHERE id=$new");
 942             }
 943             $dbh->query(sprintf("UPDATE $page_tbl SET pagename='%s' WHERE id=$id",
 944                 $dbh->escapeSimple($to)));
 945         }
 946         $this->unlock();
 947         return $id;
 948     }
 949
 950     function _update_recent_table($pageid = false)
 951     {
 952         $dbh = &$this->_dbh;
 953         extract($this->_table_names);
 954         extract($this->_expressions);
 955
 956         $pageid = (int)$pageid;
 957
 958         $this->lock();
 959         $dbh->query("DELETE FROM $recent_tbl"
 960             . ($pageid ? " WHERE id=$pageid" : ""));
 961         $dbh->query("INSERT INTO $recent_tbl"
 962             . " (id, latestversion, latestmajor, latestminor)"
 963             . " SELECT id, $maxversion, $maxmajor, $maxminor"
 964             . " FROM $version_tbl"
 965             . ($pageid ? " WHERE id=$pageid" : "")
 966             . " GROUP BY id");
 967         $this->unlock();
 968     }
 969
 970     function _update_nonempty_table($pageid = false)
 971     {
 972         $dbh = &$this->_dbh;
 973         extract($this->_table_names);
 974
 975         $pageid = (int)$pageid;
 976
 977         extract($this->_expressions);
 978         $this->lock();
 979         $dbh->query("DELETE FROM $nonempty_tbl"
 980             . ($pageid ? " WHERE id=$pageid" : ""));
 981         $dbh->query("INSERT INTO $nonempty_tbl (id)"
 982             . " SELECT $recent_tbl.id"
 983             . " FROM $recent_tbl, $version_tbl"
 984             . " WHERE $recent_tbl.id=$version_tbl.id"
 985             . "  AND version=latestversion"
 986             // We have some specifics here (Oracle)
 987             //. "  AND content<>''"
 988             . "  AND content $notempty"
 989             . ($pageid ? " AND $recent_tbl.id=$pageid" : ""));
 990         $this->unlock();
 991     }
 992
 993     /*
 994      * Grab a write lock on the tables in the SQL database.
 995      *
 996      * Calls can be nested.  The tables won't be unlocked until
 997      * _unlock_database() is called as many times as _lock_database().
 998      */
 999     public function lock($tables = array(), $write_lock = true)
1000     {
1001         if ($this->_lock_count++ == 0)
1002             $this->_lock_tables($write_lock);
1003     }
1004
1005     /*
1006      * Actually lock the required tables.
1007      */
1008     protected function _lock_tables($write_lock = true)
1009     {
1010         trigger_error("virtual", E_USER_ERROR);
1011     }
1012
1013     /**
1014      * Release a write lock on the tables in the SQL database.
1015      *
1016      * @param array $tables
1017      * @param bool $force Unlock even if not every call to lock() has been matched
1018      * by a call to unlock().
1019      *
1020      * @see _lock_database
1021      */
1022     public function unlock($tables = array(), $force = false)
1023     {
1024         if ($this->_lock_count == 0) {
1025             return;
1026         }
1027         if (--$this->_lock_count <= 0 || $force) {
1028             $this->_unlock_tables();
1029             $this->_lock_count = 0;
1030         }
1031     }
1032
1033     /**
1034      * Actually unlock the required tables.
1035      */
1036     protected function _unlock_tables()
1037     {
1038         trigger_error("virtual", E_USER_ERROR);
1039     }
1040
1041     /*
1042      * Serialize data
1043      */
1044     function _serialize($data)
1045     {
1046         if (empty($data))
1047             return '';
1048         assert(is_array($data));
1049         return serialize($data);
1050     }
1051
1052     /*
1053      * Unserialize data
1054      */
1055     function _unserialize($data)
1056     {
1057         return empty($data) ? array() : unserialize($data);
1058     }
1059
1060     /**
1061      * Callback for PEAR (DB) errors.
1062      *
1063      * @param $error PEAR_error object.
1064      */
1065     public function _pear_error_callback($error)
1066     {
1067         if ($this->_is_false_error($error))
1068             return;
1069
1070         $this->_dbh->setErrorHandling(PEAR_ERROR_PRINT); // prevent recursive loops.
1071         $this->close();
1072         trigger_error($this->_pear_error_message($error), E_USER_ERROR);
1073     }
1074
1075     /*
1076      * Detect false errors messages from PEAR DB.
1077      *
1078      * The version of PEAR DB which ships with PHP 4.0.6 has a bug in that
1079      * it doesn't recognize "LOCK" and "UNLOCK" as SQL commands which don't
1080      * return any data.  (So when a "LOCK" command doesn't return any data,
1081      * DB reports it as an error, when in fact, it's not.)
1082      *
1083      * @return bool True iff error is not really an error.
1084      */
1085     private function _is_false_error($error)
1086     {
1087         if ($error->getCode() != DB_ERROR)
1088             return false;
1089
1090         $query = $this->_dbh->last_query;
1091
1092         if (!preg_match('/^\s*"?(INSERT|UPDATE|DELETE|REPLACE|CREATE'
1093             . '|DROP|ALTER|GRANT|REVOKE|LOCK|UNLOCK)\s/', $query)
1094         ) {
1095             // Last query was not of the sort which doesn't return any data.
1096             //" <--kludge for brain-dead syntax coloring
1097             return false;
1098         }
1099
1100         if (!in_array('ismanip', get_class_methods('DB'))) {
1101             // Pear shipped with PHP 4.0.4pl1 (and before, presumably)
1102             // does not have the DB::isManip method.
1103             return true;
1104         }
1105
1106         if (DB::isManip($query)) {
1107             // If Pear thinks it's an isManip then it wouldn't have thrown
1108             // the error we're testing for....
1109             return false;
1110         }
1111
1112         return true;
1113     }
1114
1115     function _pear_error_message($error)
1116     {
1117         $class = get_class($this);
1118         $message = "$class: fatal database error\n"
1119             . "\t" . $error->getMessage() . "\n"
1120             . "\t(" . $error->getDebugInfo() . ")\n";
1121
1122         // Prevent password from being exposed during a connection error
1123         $safe_dsn = preg_replace('| ( :// .*? ) : .* (?=@) |xs',
1124             '\\1:XXXXXXXX', $this->_dsn);
1125         return str_replace($this->_dsn, $safe_dsn, $message);
1126     }
1127
1128     /*
1129      * Filter PHP errors notices from PEAR DB code.
1130      *
1131      * The PEAR DB code which ships with PHP 4.0.6 produces spurious
1132      * errors and notices.  This is an error callback (for use with
1133      * ErrorManager which will filter out those spurious messages.)
1134      * @see _is_false_error, ErrorManager
1135      */
1136     function _pear_notice_filter($err)
1137     {
1138         return ($err->isNotice()
1139             && preg_match('|DB[/\\\\]common.php$|', $err->errfile)
1140             && $err->errline == 126
1141             && preg_match('/Undefined offset: +0\b/', $err->errstr));
1142     }
1143
1144     /* some variables and functions for DB backend abstraction (action=upgrade) */
1145     function database()
1146     {
1147         return $this->_dbh->dsn['database'];
1148     }
1149
1150     function backendType()
1151     {
1152         return $this->_dbh->phptype;
1153     }
1154
1155     function connection()
1156     {
1157         return $this->_dbh->connection;
1158     }
1159
1160     function getRow($query)
1161     {
1162         return $this->_dbh->getRow($query);
1163     }
1164
1165     function listOfTables()
1166     {
1167         return $this->_dbh->getListOf('tables');
1168     }
1169
1170     function listOfFields($database, $table)
1171     {
1172         if ($this->backendType() == 'mysql') {
1173             $fields = array();
1174             assert(!empty($database));
1175             assert(!empty($table));
1176             $result = mysql_list_fields($database, $table, $this->_dbh->connection) or
1177                 trigger_error(__FILE__ . ':' . __LINE__ . ' ' . mysql_error(), E_USER_WARNING);
1178             if (!$result) return array();
1179             $columns = mysql_num_fields($result);
1180             for ($i = 0; $i < $columns; $i++) {
1181                 $fields[] = mysql_field_name($result, $i);
1182             }
1183             mysql_free_result($result);
1184             return $fields;
1185         } else {
1186             // TODO: try ADODB version?
1187             trigger_error("Unsupported dbtype and backend. Either switch to ADODB or check it manually.");
1188             return false;
1189         }
1190     }
1191 }
1192
1193 /**
1194  * This class is a generic iterator.
1195  *
1196  * WikiDB_backend_PearDB_iter only iterates over things that have
1197  * 'pagename', 'pagedata', etc. etc.
1198  *
1199  * Probably WikiDB_backend_PearDB_iter and this class should be merged
1200  * (most of the code is cut-and-paste :-( ), but I am trying to make
1201  * changes that could be merged easily.
1202  *
1203  * @author: Dan Frankowski
1204  */
1205 class WikiDB_backend_PearDB_generic_iter
1206     extends WikiDB_backend_iterator
1207 {
1208     function __construct($backend, $query_result, $field_list = NULL)
1209     {
1210         if (DB::isError($query_result)) {
1211             // This shouldn't happen, I thought.
1212             $backend->_pear_error_callback($query_result);
1213         }
1214
1215         $this->_backend = &$backend;
1216         $this->_result = $query_result;
1217         $this->_options = $field_list;
1218     }
1219
1220     function count()
1221     {
1222         if (!$this->_result) {
1223             return false;
1224         }
1225         return $this->_result->numRows();
1226     }
1227
1228     function next()
1229     {
1230         if (!$this->_result)
1231             return false;
1232
1233         $record = $this->_result->fetchRow(DB_FETCHMODE_ASSOC);
1234         if (!$record) {
1235             $this->free();
1236             return false;
1237         }
1238
1239         return $record;
1240     }
1241
1242     function reset()
1243     {
1244         if ($this->_result) {
1245             $this->_result->MoveFirst();
1246         }
1247     }
1248
1249     function free()
1250     {
1251         if ($this->_result) {
1252             $this->_result->free();
1253             $this->_result = false;
1254         }
1255     }
1256
1257     function asArray()
1258     {
1259         $result = array();
1260         while ($page = $this->next())
1261             $result[] = $page;
1262         return $result;
1263     }
1264 }
1265
1266 class WikiDB_backend_PearDB_iter
1267     extends WikiDB_backend_PearDB_generic_iter
1268 {
1269     function next()
1270     {
1271         $backend = &$this->_backend;
1272         if (!$this->_result) {
1273             return false;
1274         }
1275
1276         if (!is_object($this->_result)) {
1277             return false;
1278         }
1279
1280         $record = $this->_result->fetchRow(DB_FETCHMODE_ASSOC);
1281         if (!$record) {
1282             $this->free();
1283             return false;
1284         }
1285
1286         $pagedata = $backend->_extract_page_data($record);
1287         $rec = array('pagename' => $record['pagename'],
1288             'pagedata' => $pagedata);
1289
1290         if (!empty($record['version'])) {
1291             $rec['versiondata'] = $backend->_extract_version_data($record);
1292             $rec['version'] = $record['version'];
1293         }
1294
1295         return $rec;
1296     }
1297 }
1298
1299 class WikiDB_backend_PearDB_search extends WikiDB_backend_search_sql
1300 {
1301     // no surrounding quotes because we know it's a string
1302 }
1303
1304 // Local Variables:
1305 // mode: php
1306 // tab-width: 8
1307 // c-basic-offset: 4
1308 // c-hanging-comment-ender-p: nil
1309 // indent-tabs-mode: nil
1310 // End: