lib/WikiDB/backend/dbaBase.php

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