lib/WikiDB/backend/dbaBase.php

   1 <?php // -*-php-*-
   2 // rcs_id('$Id$');
   3
   4 require_once('lib/WikiDB/backend.php');
   5
   6 // FIXME:padding of data?  Is it needed?  dba_optimize() seems to do a good
   7 // job at packing 'gdbm' (and 'db2') databases.
   8
   9 /*
  10  * Tables:
  11  *
  12  *  page:
  13  *   Index: 'p' + pagename
  14  *  Values: latestversion . ':' . flags . ':' serialized hash of page meta data
  15  *           Currently flags = 1 if latest version has empty content.
  16  *
  17  *  version
  18  *   Index: 'v' + version:pagename
  19  *   Value: serialized hash of revision meta data, including:
  20  *          + quasi-meta-data %content
  21  *
  22  *  links
  23  *   index: 'o' + pagename
  24  *   value: serialized list of pages (names) which pagename links to.
  25  *   index: 'i' + pagename
  26  *   value: serialized list of pages which link to pagename
  27  *
  28  *  TODO:
  29  *  Don't keep tables locked the whole time.
  30  *
  31  *  More index tables:
  32  *   - Yes - RecentChanges support. Lists of most recent edits (major, minor, either).
  33  *     't' + mtime => 'a|i' + version+':'+pagename ('a': major, 'i': minor)
  34  *     Cost: Currently we have to get_all_pages and sort it by mtime.
  35  *     With a seperate t table we have to update this table on every version change.
  36  *   - No - list of pagenames for get_all_pages (very cheap: iterate page table)
  37  *   - Maybe - mostpopular list? 'h' + pagename => hits
  38   *
  39  *  Separate hit table, so we don't have to update the whole page entry
  40  *  each time we get a hit. Maybe not so important though.
  41  */
  42
  43 require_once('lib/DbaPartition.php');
  44
  45 class WikiDB_backend_dbaBase
  46 extends WikiDB_backend
  47 {
  48     function WikiDB_backend_dbaBase (&$dba) {
  49         $this->_db = &$dba;
  50         // TODO: page and version tables should be in their own files, probably.
  51         // We'll pack them all in one for now (testing).
  52         // 2004-07-09 10:07:30 rurban: It's fast enough this way.
  53         $this->_pagedb = new DbaPartition($dba, 'p');
  54         $this->_versiondb = new DbaPartition($dba, 'v');
  55         $linkdbpart = new DbaPartition($dba, 'l');
  56         $this->_linkdb = new WikiDB_backend_dbaBase_linktable($linkdbpart);
  57         $this->_dbdb = new DbaPartition($dba, 'd');
  58     }
  59
  60     function sortable_columns() {
  61         return array('pagename','mtime'/*,'author_id','author'*/);
  62     }
  63
  64     function close() {
  65         $this->_db->close();
  66     }
  67
  68     function optimize() {
  69         $this->_db->optimize();
  70     }
  71
  72     function sync() {
  73         $this->_db->sync();
  74     }
  75
  76     function rebuild($args=false) {
  77         if (!empty($args['all'])) {
  78             parent::rebuild();
  79         }
  80         // rebuild backlink table
  81         $this->_linkdb->rebuild();
  82         $this->optimize();
  83     }
  84
  85     function check($args=false) {
  86         // cleanup v?Pagename UNKNOWN0x0
  87         $errs = array();
  88         $pagedb = &$this->_pagedb;
  89         for ($page = $pagedb->firstkey();
  90              $page !== false;
  91              $page = $pagedb->nextkey())
  92         {
  93             if (!$page) {
  94                 $errs[] = "empty page $page";
  95                 trigger_error("empty page $page deleted", E_USER_WARNING);
  96                 $this->purge_page($page);
  97                 continue;
  98             }
  99             if (!($data = $pagedb->get($page))) continue;
 100             list($version,$flags,) = explode(':', $data, 3);
 101             $vdata = $this->_versiondb->get($version.":".$page);
 102             if ($vdata === false)
 103                 continue; // linkrelations
 104             // we also had for some internal version vdata is serialized strings,
 105             // need to unserialize it twice. We rather purge it.
 106             if (!is_string($vdata)
 107                 or $vdata == 'UNKNOWN\0'
 108                 or !is_array(unserialize($vdata)))
 109             {
 110                 $errs[] = "empty revision $version for $page";
 111                 trigger_error("empty revision $version for $page deleted", E_USER_WARNING);
 112                 $this->delete_versiondata($page, $version);
 113             }
 114         }
 115         // check links per default
 116         return array_merge($errs, $this->_linkdb->check());
 117     }
 118
 119     function get_pagedata($pagename) {
 120         $result = $this->_pagedb->get($pagename);
 121         if (!$result)
 122             return false;
 123         list(,,$packed) = explode(':', $result, 3);
 124         $data = unserialize($packed);
 125         return $data;
 126     }
 127
 128     function update_pagedata($pagename, $newdata) {
 129         $result = $this->_pagedb->get($pagename);
 130         if ($result) {
 131             list($latestversion,$flags,$data) = explode(':', $result, 3);
 132             $data = unserialize($data);
 133         }
 134         else {
 135             $latestversion = $flags = 0;
 136             $data = array();
 137         }
 138
 139         foreach ($newdata as $key => $val) {
 140             if (empty($val))
 141                 unset($data[$key]);
 142             else
 143                 $data[$key] = $val;
 144         }
 145         $this->_pagedb->set($pagename,
 146                             (int)$latestversion . ':'
 147                             . (int)$flags . ':'
 148                             . serialize($data));
 149     }
 150
 151     function get_latest_version($pagename) {
 152         return (int) $this->_pagedb->get($pagename);
 153     }
 154
 155     function get_previous_version($pagename, $version) {
 156         $versdb = &$this->_versiondb;
 157
 158         while (--$version > 0) {
 159             if ($versdb->exists($version . ":$pagename"))
 160                 return $version;
 161         }
 162         return false;
 163     }
 164
 165     //check $want_content
 166     function get_versiondata($pagename, $version, $want_content=false) {
 167         $data = $this->_versiondb->get((int)$version . ":$pagename");
 168         if (empty($data) or $data == 'UNKNOWN\0') return false;
 169         else {
 170             $vdata = unserialize($data);
 171             if (DEBUG and empty($vdata)) { // requires ->check
 172                 trigger_error("Delete empty revision: $pagename: ".$data, E_USER_WARNING);
 173                 $this->delete_versiondata($pagename, (int)$version);
 174             }
 175             if (!$want_content)
 176                 $vdata['%content'] = !empty($vdata['%content']);
 177             return $vdata;
 178         }
 179     }
 180
 181     /**
 182      * Can be undone and is seen in RecentChanges.
 183      * See backend.php
 184      */
 185     function delete_page($pagename) {
 186         $version = $this->get_latest_version($pagename);
 187         $data = $this->_versiondb->get((int)$version . ":$pagename");
 188         // fix broken pages
 189         if (!is_array($data) or empty($data)) {
 190             if (is_string($data) and ($vdata = @unserialize($data))) {
 191                 trigger_error("Fixed broken page version $pagename. Run 'Check WikiDB'",
 192                               E_USER_NOTICE);
 193                 $data = $vdata;
 194             } else // already empty page
 195                 $data = array();
 196         }
 197         assert(is_array($data) and !empty($data)); // mtime
 198         $data['%content'] = '';
 199         $data['mtime'] = time();
 200         $data['summary'] = "removed by ".$GLOBALS["request"]->_deduceUsername();
 201         $this->set_versiondata($pagename, $version+1, $data);
 202         $this->set_links($pagename, false);
 203     }
 204
 205     /**
 206      * Completely delete all page revisions from the database.
 207      */
 208     function purge_page($pagename) {
 209         $pagedb = &$this->_pagedb;
 210         $versdb = &$this->_versiondb;
 211
 212         $version = $this->get_latest_version($pagename);
 213         while ($version > 0) {
 214             $versdb->set($version-- . ":$pagename", false);
 215         }
 216         $pagedb->set($pagename, false);
 217
 218         $this->set_links($pagename, false);
 219     }
 220
 221     function rename_page($pagename, $to) {
 222         $result = $this->_pagedb->get($pagename);
 223         if ($result) {
 224             list($version, $flags, $data) = explode(':', $result, 3);
 225             $data = unserialize($data);
 226         }
 227         else
 228             return false;
 229
 230         $links = $this->_linkdb->get_links($pagename, false, false);
 231         $data['pagename'] = $to;
 232         $this->_pagedb->set($to,
 233                             (int)$version . ':'
 234                             . (int)$flags . ':'
 235                             . serialize($data));
 236         // move over the latest version only
 237         $pvdata = $this->get_versiondata($pagename, $version, true);
 238         $data['mtime'] = time();
 239         $data['summary'] = "renamed from ".$pagename
 240                           ." by ".$GLOBALS["request"]->_deduceUsername();
 241         $this->set_versiondata($to, $version, $pvdata);
 242
 243         // update links and backlinks
 244         $this->_linkdb->set_links($to, $links);
 245         // better: update all back-/inlinks for all outlinks.
 246
 247         $this->_pagedb->delete($pagename);
 248         return true;
 249     }
 250
 251     /**
 252      * Delete an old revision of a page.
 253      */
 254     function delete_versiondata($pagename, $version) {
 255         $versdb = &$this->_versiondb;
 256
 257         $latest = $this->get_latest_version($pagename);
 258
 259         assert($version > 0);
 260         assert($version <= $latest);
 261
 262         $versdb->set((int)$version . ":$pagename", false);
 263
 264         if ($version == $latest) {
 265             $previous = $this->get_previous_version($pagename, $version);
 266             if ($previous > 0) {
 267                 $pvdata = $this->get_versiondata($pagename, $previous);
 268                 $is_empty = empty($pvdata['%content']);
 269             }
 270             else
 271                 $is_empty = true;
 272             $this->_update_latest_version($pagename, $previous, $is_empty);
 273         }
 274     }
 275
 276     /**
 277      * Create a new revision of a page.
 278      */
 279     function set_versiondata($pagename, $version, $data) {
 280         $versdb = &$this->_versiondb;
 281         // fix broken pages
 282         if (!is_array($data) or empty($data)) {
 283             if (is_string($data) and ($vdata = @unserialize($data))) {
 284                 trigger_error("broken page version $pagename. Run Check WikiDB",
 285                               E_USER_NOTICE);
 286                 $data = $vdata;
 287             } else
 288                 $data = array();
 289         }
 290         assert(is_array($data) and !empty($data)); // mtime
 291         $versdb->set((int)$version . ":$pagename", serialize($data));
 292         if ($version > $this->get_latest_version($pagename))
 293             $this->_update_latest_version($pagename, $version, empty($data['%content']));
 294     }
 295
 296     function _update_latest_version($pagename, $latest, $flags) {
 297         $pagedb = &$this->_pagedb;
 298
 299         $pdata = $pagedb->get($pagename);
 300         if ($pdata)
 301             list(,,$pagedata) = explode(':',$pdata,3);
 302         else
 303             $pagedata = serialize(array());
 304
 305         $pagedb->set($pagename, (int)$latest . ':' . (int)$flags . ":$pagedata");
 306     }
 307
 308     function numPages($include_empty=false, $exclude='') {
 309         $pagedb = &$this->_pagedb;
 310         $count = 0;
 311         for ($page = $pagedb->firstkey(); $page!== false; $page = $pagedb->nextkey()) {
 312             if (!$page) {
 313                 assert(!empty($page));
 314                 continue;
 315             }
 316             if ($exclude and in_array($page, $exclude)) continue;
 317             if (!$include_empty) {
 318                 if (!($data = $pagedb->get($page))) continue;
 319                 list($latestversion,$flags,) = explode(':', $data, 3);
 320                 unset($data);
 321                 if ($latestversion == 0 || $flags != 0)
 322                     continue;   // current content is empty
 323             }
 324             $count++;
 325         }
 326         return $count;
 327     }
 328
 329     function get_all_pages($include_empty=false, $sortby='', $limit='', $exclude='') {
 330         $pagedb = &$this->_pagedb;
 331         $pages = array();
 332         $from = 0; $i = 0; $count = 0;
 333         if ($limit) { // extract from,count from limit
 334             list($from,$count) = $this->limit($limit);
 335         }
 336         for ($page = $pagedb->firstkey(); $page!== false; $page = $pagedb->nextkey()) {
 337             if (!$page) {
 338                 assert(!empty($page));
 339                 continue;
 340             }
 341             if ($exclude and in_array($page, $exclude)) continue;
 342             if ($limit and $from) {
 343                 $i++;
 344                 if ($i < $from) continue;
 345             }
 346             if ($limit and count($pages) >= $count) break;
 347             if (!$include_empty) {
 348                 if (!($data = $pagedb->get($page))) continue;
 349                 list($latestversion,$flags,) = explode(':', $data, 3);
 350                 unset($data);
 351                 if ($latestversion == 0 || $flags != 0)
 352                     continue;   // current content is empty
 353             }
 354             $pages[] = $page;
 355         }
 356         return new WikiDB_backend_dbaBase_pageiter
 357             ($this, $pages,
 358              array('sortby'=>$sortby/*,
 359                    'limit' =>$limit*/));
 360     }
 361
 362     function set_links($pagename, $links) {
 363         $this->_linkdb->set_links($pagename, $links);
 364     }
 365
 366     function get_links($pagename, $reversed=true, $include_empty=false,
 367                        $sortby='', $limit='', $exclude='',
 368                        $want_relations=false)
 369     {
 370         // optimization: if no relation at all is found, mark it in the iterator.
 371         $links = $this->_linkdb->get_links($pagename, $reversed, $want_relations);
 372
 373         return new WikiDB_backend_dbaBase_pageiter
 374             ($this, $links,
 375              array('sortby'=>$sortby,
 376                    'limit' =>$limit,
 377                    'exclude'=>$exclude,
 378                    'want_relations'=>$want_relations,
 379                    'found_relations' => $want_relations
 380                      ? $this->_linkdb->found_relations : 0
 381                    ));
 382     }
 383
 384     /**
 385      * @access public
 386      *
 387      * @return array of all linkrelations
 388      * Faster than the dumb WikiDB method.
 389      */
 390     function list_relations($also_attributes=false,
 391                             $only_attributes=false,
 392                             $sorted=true)
 393     {
 394         $linkdb = &$this->_linkdb;
 395         $relations = array();
 396         for ($link = $linkdb->_db->firstkey();
 397              $link!== false;
 398              $link = $linkdb->_db->nextkey())
 399         {
 400             if ($link[0] != 'o') continue;
 401             $links = $linkdb->_get_links('o', substr($link,1));
 402             foreach ($links as $link) { // linkto => page, linkrelation => page
 403                 if (is_array($link)
 404                     and $link['relation']
 405                     and !in_array($link['relation'], $relations))
 406                 {
 407                     $is_attribute = empty($link['linkto']); // a relation has both
 408                     if ($is_attribute) {
 409                         if ($only_attributes or $also_attributes)
 410                             $relations[] = $link['relation'];
 411                     } elseif (!$only_attributes) {
 412                           $relations[] = $link['relation'];
 413                     }
 414                 }
 415             }
 416         }
 417         if ($sorted) {
 418             sort($relations);
 419             reset($relations);
 420         }
 421         return $relations;
 422     }
 423
 424     /**
 425      * WikiDB_backend_dumb_LinkSearchIter searches over all
 426      * pages and then all its links.  Since there are less
 427      * links than pages, and we easily get the pagename from
 428      * the link key, we iterate here directly over the
 429      * linkdb and check the pagematch there.
 430      *
 431      * @param $pages     object A TextSearchQuery object for the pagename filter.
 432      * @param $query     object A SearchQuery object (Text or Numeric) for the linkvalues,
 433      *                          linkto, linkfrom (=backlink), relation or attribute values.
 434      * @param $linktype  string One of the 4 linktypes "linkto",
 435      *                          "linkfrom" (=backlink), "relation" or "attribute".
 436      *                          Maybe also "relation+attribute" for the advanced search.
 437      * @param $relation  object A TextSearchQuery for the linkname or false.
 438      * @param $options   array Currently ignored. hash of sortby, limit, exclude.
 439      * @return object A WikiDB_backend_iterator.
 440      * @see WikiDB::linkSearch
 441      */
 442     function link_search( $pages, $query, $linktype,
 443                           $relation=false, $options=array() )
 444     {
 445         $linkdb = &$this->_linkdb;
 446         $links = array();
 447         $reverse = false;
 448         $want_relations = false;
 449         if ($linktype == 'relation') {
 450             $want_relations = true;
 451             $field = 'linkrelation';
 452         }
 453         if ($linktype == 'attribute') {
 454             $want_relations = true;
 455             $field = 'attribute';
 456         }
 457         if ($linktype == 'linkfrom') {
 458             $reverse = true;
 459         }
 460
 461         for ($link = $linkdb->_db->firstkey();
 462              $link!== false;
 463              $link = $linkdb->_db->nextkey())
 464         {
 465             $type = $reverse ? 'i' : 'o';
 466             if ($link[0] != $type) continue;
 467             $pagename = substr($link, 1);
 468             if (!$pages->match($pagename)) continue;
 469             if ($linktype == 'attribute') {
 470                 $page = $GLOBALS['request']->_dbi->getPage($pagename);
 471                 $attribs = $page->get('attributes');
 472                 if ($attribs) {
 473                     /* Optimization on expressive searches:
 474                        for queries with multiple attributes.
 475                        Just take the defined placeholders from the query(ies)
 476                        if there are more attributes than query variables.
 477                     */
 478                     if ($query->getType() != 'text'
 479                         and !$relation
 480                         and ((count($vars = $query->getVars()) > 1)
 481                              or (count($attribs) > count($vars))))
 482                     {
 483                         // names must strictly match. no * allowed
 484                         if (!$query->can_match($attribs)) continue;
 485                         if (!($result = $query->match($attribs))) continue;
 486                         foreach ($result as $r) {
 487                             $r['pagename'] = $pagename;
 488                             $links[] = $r;
 489                         }
 490                     } else {
 491                         // textsearch or simple value. no strict bind by name needed
 492                         foreach ($attribs as $attribute => $value) {
 493                             if ($relation and !$relation->match($attribute)) continue;
 494                             if (!$query->match($value)) continue;
 495                             $links[] = array('pagename'  => $pagename,
 496                                              'linkname'  => $attribute,
 497                                              'linkvalue' => $value);
 498                         }
 499                     }
 500                 }
 501             }
 502             else {
 503                 // TODO: honor limits. this can get large.
 504                 if ($want_relations) {
 505                     // MAP linkrelation : pagename => thispagename : linkname : linkvalue
 506                     $_links = $linkdb->_get_links('o', $pagename);
 507                     foreach ($_links as $link) { // linkto => page, linkrelation => page
 508                         if (!isset($link['relation']) or !$link['relation']) continue;
 509                         if ($relation and !$relation->match($link['relation'])) continue;
 510                         if (!$query->match($link['linkto'])) continue;
 511                         $links[] = array('pagename'  => $pagename,
 512                                          'linkname'  => $link['relation'],
 513                                          'linkvalue' => $link['linkto']);
 514                     }
 515                 } else {
 516                     $_links = $linkdb->_get_links($reverse ? 'i' : 'o', $pagename);
 517                     foreach ($_links as $link) { // linkto => page
 518                         if (is_array($link))
 519                             $link = $link['linkto'];
 520                         if (!$query->match($link)) continue;
 521                         $links[] = array('pagename'  => $pagename,
 522                                          'linkname'  => '',
 523                                          'linkvalue' => $link);
 524                     }
 525                 }
 526             }
 527         }
 528         $options['want_relations'] = true; // Iter hack to force return of the whole hash
 529         return new WikiDB_backend_dbaBase_pageiter($this, $links, $options);
 530     }
 531
 532     /**
 533      * Handle multi-searches for many relations and attributes in one expression.
 534      * Bind all required attributes and relations per page together and pass it
 535      * to one query.
 536      *   (is_a::city and population < 20000) and (*::city and area > 1000000)
 537      *   (is_a::city or linkto::CategoryCountry) and population < 20000 and area > 1000000
 538      * Note that the 'linkto' and 'linkfrom' links are relations, containing an array.
 539      *
 540      * @param $pages     object A TextSearchQuery object for the pagename filter.
 541      * @param $query     object A SemanticSearchQuery object for the links.
 542      * @param $options   array  Currently ignored. hash of sortby, limit, exclude
 543      *                          for the pagelist.
 544      * @return object A WikiDB_backend_iterator.
 545      * @see WikiDB::linkSearch
 546      */
 547     function relation_search( $pages, $query, $options=array() ) {
 548         $linkdb = &$this->_linkdb;
 549         $links = array();
 550         // We need to detect which attributes and relation names we should look for. NYI
 551         $want_attributes = $query->hasAttributes();
 552         $want_relation = $query->hasRelations();
 553         $linknames = $query->getLinkNames();
 554         // create a hash for faster checks
 555         $linkcheck = array();
 556         foreach ($linknames as $l) $linkcheck[$l] = 1;
 557
 558         for ($link = $linkdb->_db->firstkey();
 559              $link!== false;
 560              $link = $linkdb->_db->nextkey())
 561         {
 562             $type = $reverse ? 'i' : 'o';
 563             if ($link[0] != $type) continue;
 564             $pagename = substr($link, 1);
 565             if (!$pages->match($pagename)) continue;
 566             $pagelinks = array();
 567             if ($want_attributes) {
 568                 $page = $GLOBALS['request']->_dbi->getPage($pagename);
 569                 $attribs = $page->get('attributes');
 570                 $pagelinks = $attribs;
 571             }
 572             if ($want_relations) {
 573                 // all links contain arrays of pagenames, just the attributes
 574                 // are guaranteed to be singular
 575                 if (isset($linkcheck['linkfrom'])) {
 576                     $pagelinks['linkfrom'] = $linkdb->_get_links('i', $pagename);
 577                 }
 578                 $outlinks = $linkdb->_get_links('o', $pagename);
 579                 $want_to = isset($linkcheck['linkto']);
 580                 foreach ($outlinks as $link) { // linkto => page, relation => page
 581                     // all named links
 582                     if ((isset($link['relation'])) and $link['relation']
 583                         and isset($linkcheck[$link['relation']]))
 584                         $pagelinks[$link['relation']][] = $link['linkto'];
 585                     if ($want_to)
 586                         $pagelinks['linkto'][] = is_array($link) ? $link['linkto'] : $link;
 587                 }
 588             }
 589             if ($result = $query->match($pagelinks)) {
 590                 $links = array_merge($links, $result);
 591             }
 592         }
 593         $options['want_relations'] = true; // Iter hack to force return of the whole hash
 594         return new WikiDB_backend_dbaBase_pageiter($this, $links, $options);
 595     }
 596 };
 597
 598 function WikiDB_backend_dbaBase_sortby_pagename_ASC ($a, $b) {
 599     return strcasecmp($a, $b);
 600 }
 601 function WikiDB_backend_dbaBase_sortby_pagename_DESC ($a, $b) {
 602     return strcasecmp($b, $a);
 603 }
 604 function WikiDB_backend_dbaBase_sortby_mtime_ASC ($a, $b) {
 605     return WikiDB_backend_dbaBase_sortby_num($a, $b, 'mtime');
 606 }
 607 function WikiDB_backend_dbaBase_sortby_mtime_DESC ($a, $b) {
 608     return WikiDB_backend_dbaBase_sortby_num($b, $a, 'mtime');
 609 }
 610 /*
 611 function WikiDB_backend_dbaBase_sortby_hits_ASC ($a, $b) {
 612     return WikiDB_backend_dbaBase_sortby_num($a, $b, 'hits');
 613 }
 614 function WikiDB_backend_dbaBase_sortby_hits_DESC ($a, $b) {
 615     return WikiDB_backend_dbaBase_sortby_num($b, $a, 'hits');
 616 }
 617 */
 618 function WikiDB_backend_dbaBase_sortby_num($aname, $bname, $field) {
 619     global $request;
 620     $dbi = $request->getDbh();
 621     // fields are stored in versiondata
 622     $av = $dbi->_backend->get_latest_version($aname);
 623     $bv = $dbi->_backend->get_latest_version($bname);
 624     $a = $dbi->_backend->get_versiondata($aname, $av, false);
 625     if (!$a) return -1;
 626     $b = $dbi->_backend->get_versiondata($bname, $bv, false);
 627     if (!$b or !isset($b[$field])) return 0;
 628     if (empty($a[$field])) return -1;
 629     if ((!isset($a[$field]) and !isset($b[$field])) or ($a[$field] === $b[$field])) {
 630         return 0;
 631     } else {
 632         return ($a[$field] < $b[$field]) ? -1 : 1;
 633     }
 634 }
 635
 636 class WikiDB_backend_dbaBase_pageiter
 637 extends WikiDB_backend_iterator
 638 {
 639     // fixed for linkrelations
 640     function WikiDB_backend_dbaBase_pageiter(&$backend, &$pages, $options=false) {
 641         $this->_backend = $backend;
 642         $this->_options = $options;
 643         if ($pages) {
 644             if (!empty($options['sortby'])) {
 645                 $sortby = WikiDB_backend::sortby($options['sortby'], 'db',
 646                                                  array('pagename','mtime'));
 647                 // check for which column to sortby
 648                 if ($sortby and !strstr($sortby, "hits ")) {
 649                     usort($pages, 'WikiDB_backend_dbaBase_sortby_'
 650                           .str_replace(' ','_',$sortby));
 651                 }
 652             }
 653             if (!empty($options['limit'])) {
 654                 list($offset,$limit) = WikiDB_backend::limit($options['limit']);
 655                 $pages = array_slice($pages, $offset, $limit);
 656             }
 657             $this->_pages = $pages;
 658         } else
 659             $this->_pages = array();
 660     }
 661
 662     // fixed for relations
 663     function next() {
 664         if ( ! ($page = array_shift($this->_pages)) )
 665             return false;
 666         if (!empty($this->_options['want_relations'])) {
 667             // $linkrelation = $page['linkrelation'];
 668             $pagename = $page['pagename'];
 669             if (!empty($this->_options['exclude'])
 670                 and in_array($pagename, $this->_options['exclude']))
 671                 return $this->next();
 672             return $page;
 673         }
 674         if (!empty($this->_options['exclude'])
 675             and in_array($page, $this->_options['exclude']))
 676             return $this->next();
 677         return array('pagename' => $page);
 678     }
 679
 680     function reset() {
 681         reset($this->_pages);
 682     }
 683     function free() {
 684         $this->_pages = array();
 685     }
 686 };
 687
 688 class WikiDB_backend_dbaBase_linktable
 689 {
 690     function WikiDB_backend_dbaBase_linktable(&$dba) {
 691         $this->_db = &$dba;
 692     }
 693
 694     //TODO: try storing link lists as hashes rather than arrays.
 695     //      backlink deletion would be faster.
 696     function get_links($page, $reversed=true, $want_relations=false) {
 697         if ($want_relations) {
 698             $this->found_relations = 0;
 699             $links = $this->_get_links($reversed ? 'i' : 'o', $page);
 700             $linksonly = array();
 701             foreach ($links as $link) { // linkto => page, linkrelation => page
 702                 if (is_array($link) and isset($link['relation'])) {
 703                     if ($link['relation'])
 704                         $this->found_relations++;
 705                     $linksonly[] = array('pagename'     => $link['linkto'],
 706                                          'linkrelation' => $link['relation']);
 707                 } else { // empty relations are stripped
 708                     $linksonly[] = array('pagename' => $link['linkto']);
 709                 }
 710             }
 711             return $linksonly;
 712         } else {
 713             $links = $this->_get_links($reversed ? 'i' : 'o', $page);
 714             $linksonly = array();
 715             foreach ($links as $link) {
 716                 if (is_array($link)) {
 717                     $linksonly[] = $link['linkto'];
 718                 } else
 719                     $linksonly[] = $link;
 720             }
 721             return $linksonly;
 722         }
 723     }
 724
 725     // fixed: relations ready
 726     function set_links($page, $links) {
 727
 728         $oldlinks = $this->get_links($page, false, false);
 729
 730         if (!is_array($links)) {
 731             assert(empty($links));
 732             $links = array();
 733         }
 734         $this->_set_links('o', $page, $links);
 735
 736         /* Now for the backlink update we squash the linkto hashes into a simple array */
 737         $newlinks = array();
 738         foreach ($links as $hash) {
 739             if (!empty($hash['linkto']) and !in_array($hash['linkto'], $newlinks))
 740                  // for attributes it's empty
 741                 $newlinks[] = $hash['linkto'];
 742             elseif (is_string($hash) and !in_array($hash, $newlinks))
 743                 $newlinks[] = $hash;
 744         }
 745         //$newlinks = array_unique($newlinks);
 746         sort($oldlinks);
 747         sort($newlinks);
 748
 749         reset($newlinks);
 750         reset($oldlinks);
 751         $new = current($newlinks);
 752         $old = current($oldlinks);
 753         while ($new !== false || $old !== false) {
 754             if ($old === false || ($new !== false && $new < $old)) {
 755                 // $new is a new link (not in $oldlinks).
 756                 $this->_add_backlink($new, $page);
 757                 $new = next($newlinks);
 758             }
 759             elseif ($new === false || $old < $new) {
 760                 // $old is a obsolete link (not in $newlinks).
 761                 $this->_delete_backlink($old, $page);
 762                 $old = next($oldlinks);
 763             }
 764             else {
 765                 // Unchanged link (in both $newlist and $oldlinks).
 766                 assert($new == $old);
 767                 $new = next($newlinks);
 768                 $old = next($oldlinks);
 769             }
 770         }
 771     }
 772
 773     /**
 774      * Rebuild the back-link index.
 775      *
 776      * This should never be needed, but if the database gets hosed for some reason,
 777      * this should put it back into a consistent state.
 778      *
 779      * We assume the forward links in the our table are correct, and recalculate
 780      * all the backlinks appropriately.
 781      */
 782     function rebuild () {
 783         $db = &$this->_db;
 784
 785         // Delete the backlink tables, make a list of lo.page keys.
 786         $okeys = array();
 787         for ($key = $db->firstkey(); $key; $key = $db->nextkey()) {
 788             if ($key[0] == 'i')
 789                 $db->delete($key);
 790             elseif ($key[0] == 'o')
 791                 $okeys[] = $key;
 792             else {
 793                 trigger_error("Bad key in linktable: '$key'", E_USER_WARNING);
 794             $db->delete($key);
 795         }
 796         }
 797         foreach ($okeys as $key) {
 798             $page = substr($key,1);
 799             $links = $this->_get_links('o', $page);
 800             $db->delete($key);
 801             $this->set_links($page, $links);
 802         }
 803     }
 804
 805     function check() {
 806         $db = &$this->_db;
 807
 808         // FIXME: check for sortedness and uniqueness in links lists.
 809
 810         for ($key = $db->firstkey(); $key; $key = $db->nextkey()) {
 811             if (strlen($key) < 1 || ($key[0] != 'i' && $key[0] != 'o')) {
 812                 $errs[] = "Bad key '$key' in table";
 813                 continue;
 814             }
 815             $page = substr($key, 1);
 816             if ($key[0] == 'o') {
 817                 // Forward links.
 818                 foreach($this->_get_links('o', $page) as $link) {
 819                     $link = $link['linkto'];
 820                     if (!$this->_has_link('i', $link, $page))
 821                         $errs[] = "backlink entry missing for link '$page'->'$link'";
 822                 }
 823             }
 824             else {
 825                 assert($key[0] == 'i');
 826                 // Backlinks.
 827                 foreach($this->_get_links('i', $page) as $link) {
 828                     if (!$this->_has_link('o', $link, $page))
 829                         $errs[] = "link entry missing for backlink '$page'<-'$link'";
 830                 }
 831             }
 832         }
 833         //if ($errs) $this->rebuild();
 834         return isset($errs) ? $errs : false;
 835     }
 836
 837     /* TODO: Add another lrRelationName key for relations.
 838      * lrRelationName: frompage => topage
 839      */
 840
 841     function _add_relation($page, $linkedfrom) {
 842         $relations = $this->_get_links('r', $page);
 843         $backlinks[] = $linkedfrom;
 844         sort($backlinks);
 845         $this->_set_links('i', $page, $backlinks);
 846     }
 847
 848     function _add_backlink($page, $linkedfrom) {
 849         $backlinks = $this->_get_links('i', $page);
 850         $backlinks[] = $linkedfrom;
 851         sort($backlinks);
 852         $this->_set_links('i', $page, $backlinks);
 853     }
 854
 855     function _delete_backlink($page, $linkedfrom) {
 856         $backlinks = $this->_get_links('i', $page);
 857         foreach ($backlinks as $key => $backlink) {
 858             if ($backlink == $linkedfrom)
 859                 unset($backlinks[$key]);
 860         }
 861         $this->_set_links('i', $page, $backlinks);
 862     }
 863
 864     function _has_link($which, $page, $link) {
 865         $links = $this->_get_links($which, $page);
 866         // since links are always sorted, break if >
 867         // TODO: binary search
 868         foreach($links as $l) {
 869             if ($l['linkto'] == $link)
 870                 return true;
 871             if ($l['linkto'] > $link)
 872                 return false;
 873         }
 874         return false;
 875     }
 876
 877     function _get_links($which, $page) {
 878         $data = $this->_db->get($which . $page);
 879         return $data ? unserialize($data) : array();
 880     }
 881
 882     function _set_links($which, $page, &$links) {
 883         $key = $which . $page;
 884         if ($links)
 885             $this->_db->set($key, serialize($links));
 886         else
 887             $this->_db->set($key, false);
 888     }
 889 }
 890
 891 // (c-file-style: "gnu")
 892 // Local Variables:
 893 // mode: php
 894 // tab-width: 8
 895 // c-basic-offset: 4
 896 // c-hanging-comment-ender-p: nil
 897 // indent-tabs-mode: nil
 898 // End:
 899 ?>