1<?php
2
3namespace dokuwiki\plugin\struct\meta;
4
5use dokuwiki\plugin\struct\types\AutoSummary;
6use dokuwiki\plugin\struct\types\DateTime;
7use dokuwiki\plugin\struct\types\Decimal;
8use dokuwiki\plugin\struct\types\Page;
9use dokuwiki\plugin\struct\types\User;
10
11class Search
12{
13    /**
14     * This separator will be used to concat multi values to flatten them in the result set
15     */
16    public const CONCAT_SEPARATOR = "\n!_-_-_-_-_!\n";
17
18    /**
19     * The list of known and allowed comparators
20     * (order matters)
21     */
22    public static $COMPARATORS = array(
23        '<=', '>=', '=*', '=', '<', '>', '!=', '!~', '~', ' IN '
24    );
25
26    /** @var  \helper_plugin_sqlite */
27    protected $sqlite;
28
29    /** @var Schema[] list of schemas to query */
30    protected $schemas = array();
31
32    /** @var Column[] list of columns to select */
33    protected $columns = array();
34
35    /** @var array the sorting of the result */
36    protected $sortby = array();
37
38    /** @var array the filters */
39    protected $filter = array();
40
41    /** @var array list of aliases tables can be referenced by */
42    protected $aliases = array();
43
44    /** @var  int begin results from here */
45    protected $range_begin = 0;
46
47    /** @var  int end results here */
48    protected $range_end = 0;
49
50    /** @var int the number of results */
51    protected $count = -1;
52    /** @var  string[] the PIDs of the result rows */
53    protected $result_pids = null;
54    /** @var  array the row ids of the result rows */
55    protected $result_rids = [];
56    /** @var  array the revisions of the result rows */
57    protected $result_revs = [];
58
59    /**
60     * Search constructor.
61     */
62    public function __construct()
63    {
64        /** @var \helper_plugin_struct_db $plugin */
65        $plugin = plugin_load('helper', 'struct_db');
66        $this->sqlite = $plugin->getDB();
67    }
68
69    /**
70     * Add a schema to be searched
71     *
72     * Call multiple times for multiple schemas.
73     *
74     * @param string $table
75     * @param string $alias
76     */
77    public function addSchema($table, $alias = '')
78    {
79        $schema = new Schema($table);
80        if (!$schema->getId()) {
81            throw new StructException('schema missing', $table);
82        }
83
84        $this->schemas[$schema->getTable()] = $schema;
85        if ($alias) $this->aliases[$alias] = $schema->getTable();
86    }
87
88    /**
89     * Add a column to be returned by the search
90     *
91     * Call multiple times for multiple columns. Be sure the referenced tables have been
92     * added before
93     *
94     * @param string $colname may contain an alias
95     */
96    public function addColumn($colname)
97    {
98        if ($this->processWildcard($colname)) return; // wildcard?
99        if ($colname[0] == '-') { // remove column from previous wildcard lookup
100            $colname = substr($colname, 1);
101            foreach ($this->columns as $key => $col) {
102                if ($col->getLabel() == $colname) unset($this->columns[$key]);
103            }
104            return;
105        }
106
107        $col = $this->findColumn($colname);
108        if (!$col) return; //FIXME do we really want to ignore missing columns?
109        $this->columns[] = $col;
110    }
111
112    /**
113     * Add sorting options
114     *
115     * Call multiple times for multiple columns. Be sure the referenced tables have been
116     * added before
117     *
118     * @param string $colname may contain an alias
119     * @param bool $asc sort direction (ASC = true, DESC = false)
120     * @param bool $nc set true for caseinsensitivity
121     */
122    public function addSort($colname, $asc = true, $nc = true)
123    {
124        $col = $this->findColumn($colname);
125        if (!$col) return; //FIXME do we really want to ignore missing columns?
126
127        $this->sortby[$col->getFullQualifiedLabel()] = array($col, $asc, $nc);
128    }
129
130    /**
131     * Returns all set sort columns
132     *
133     * @return array
134     */
135    public function getSorts()
136    {
137        return $this->sortby;
138    }
139
140    /**
141     * Adds a filter
142     *
143     * @param string $colname may contain an alias
144     * @param string|string[] $value
145     * @param string $comp @see self::COMPARATORS
146     * @param string $op either 'OR' or 'AND'
147     */
148    public function addFilter($colname, $value, $comp, $op = 'OR')
149    {
150        /* Convert certain filters into others
151         * this reduces the number of supported filters to implement in types */
152        if ($comp == '*~') {
153            $value = $this->filterWrapAsterisks($value);
154            $comp = '~';
155        } elseif ($comp == '<>') {
156            $comp = '!=';
157        }
158
159        if (!in_array($comp, self::$COMPARATORS))
160            throw new StructException("Bad comperator. Use " . join(',', self::$COMPARATORS));
161        if ($op != 'OR' && $op != 'AND')
162            throw new StructException('Bad filter type . Only AND or OR allowed');
163
164        $col = $this->findColumn($colname);
165        if (!$col) return; // ignore missing columns, filter might have been for different schema
166
167        // map filter operators to SQL syntax
168        switch ($comp) {
169            case '~':
170                $comp = 'LIKE';
171                break;
172            case '!~':
173                $comp = 'NOT LIKE';
174                break;
175            case '=*':
176                $comp = 'REGEXP';
177                break;
178        }
179
180        // we use asterisks, but SQL wants percents
181        if ($comp == 'LIKE' || $comp == 'NOT LIKE') {
182            $value = $this->filterChangeToLike($value);
183        }
184
185        if ($comp == ' IN ' && !is_array($value)) {
186            $value = $this->parseFilterValueList($value);
187            //col IN ('a', 'b', 'c') is equal to col = 'a' OR 'col = 'b' OR col = 'c'
188            $comp = '=';
189        }
190
191        // add the filter
192        $this->filter[] = array($col, $value, $comp, $op);
193    }
194
195    /**
196     * Parse SQLite row value into array
197     *
198     * @param string $value
199     * @return string[]
200     */
201    protected function parseFilterValueList($value)
202    {
203        $Handler = new FilterValueListHandler();
204        $LexerClass = class_exists('\Doku_Lexer') ? '\Doku_Lexer' : '\dokuwiki\Parsing\Lexer\Lexer';
205        $isLegacy = $LexerClass === '\Doku_Lexer';
206        /** @var \Doku_Lexer|\dokuwiki\Parsing\Lexer\Lexer $Lexer */
207        $Lexer = new $LexerClass($Handler, 'base', true);
208
209
210        $Lexer->addEntryPattern('\(', 'base', 'row');
211        $Lexer->addPattern('\s*,\s*', 'row');
212        $Lexer->addExitPattern('\)', 'row');
213
214        $Lexer->addEntryPattern('"', 'row', 'double_quote_string');
215        $Lexer->addSpecialPattern('\\\\"', 'double_quote_string', 'escapeSequence');
216        $Lexer->addExitPattern('"', 'double_quote_string');
217
218        $Lexer->addEntryPattern("'", 'row', 'singleQuoteString');
219        $Lexer->addSpecialPattern("\\\\'", 'singleQuoteString', 'escapeSequence');
220        $Lexer->addExitPattern("'", 'singleQuoteString');
221
222        $Lexer->mapHandler('double_quote_string', 'singleQuoteString');
223
224        $Lexer->addSpecialPattern('[-+]?[0-9]*\.?[0-9]+(?:[eE][-+]?[0-9]+)?', 'row', 'number');
225
226        $res = $Lexer->parse($value);
227
228        $currentMode = $isLegacy ? $Lexer->_mode->getCurrent() : $Lexer->getModeStack()->getCurrent();
229        if (!$res || $currentMode != 'base') {
230            throw new StructException('invalid row value syntax');
231        }
232
233        return $Handler->getRow();
234    }
235
236    /**
237     * Wrap given value in asterisks
238     *
239     * @param string|string[] $value
240     * @return string|string[]
241     */
242    protected function filterWrapAsterisks($value)
243    {
244        $map = function ($input) {
245            return "*$input*";
246        };
247
248        if (is_array($value)) {
249            $value = array_map($map, $value);
250        } else {
251            $value = $map($value);
252        }
253        return $value;
254    }
255
256    /**
257     * Change given string to use % instead of *
258     *
259     * @param string|string[] $value
260     * @return string|string[]
261     */
262    protected function filterChangeToLike($value)
263    {
264        $map = function ($input) {
265            return str_replace('*', '%', $input);
266        };
267
268        if (is_array($value)) {
269            $value = array_map($map, $value);
270        } else {
271            $value = $map($value);
272        }
273        return $value;
274    }
275
276    /**
277     * Set offset for the results
278     *
279     * @param int $offset
280     */
281    public function setOffset($offset)
282    {
283        $limit = 0;
284        if ($this->range_end) {
285            // if there was a limit set previously, the range_end needs to be recalculated
286            $limit = $this->range_end - $this->range_begin;
287        }
288        $this->range_begin = $offset;
289        if ($limit) $this->setLimit($limit);
290    }
291
292    /**
293     * Limit results to this number
294     *
295     * @param int $limit Set to 0 to disable limit again
296     */
297    public function setLimit($limit)
298    {
299        if ($limit) {
300            $this->range_end = $this->range_begin + $limit;
301        } else {
302            $this->range_end = 0;
303        }
304    }
305
306    /**
307     * Return the number of results (regardless of limit and offset settings)
308     *
309     * Use this to implement paging. Important: this may only be called after running @return int
310     * @see execute()
311     *
312     */
313    public function getCount()
314    {
315        if ($this->count < 0) throw new StructException('Count is only accessible after executing the search');
316        return $this->count;
317    }
318
319    /**
320     * Returns the PID associated with each result row
321     *
322     * Important: this may only be called after running @return \string[]
323     * @see execute()
324     *
325     */
326    public function getPids()
327    {
328        if ($this->result_pids === null)
329            throw new StructException('PIDs are only accessible after executing the search');
330        return $this->result_pids;
331    }
332
333    /**
334     * Returns the rid associated with each result row
335     *
336     * Important: this may only be called after running @return array
337     * @see execute()
338     *
339     */
340    public function getRids()
341    {
342        if ($this->result_rids === null)
343            throw new StructException('rids are only accessible after executing the search');
344        return $this->result_rids;
345    }
346
347    /**
348     * Returns the rid associated with each result row
349     *
350     * Important: this may only be called after running @return array
351     * @see execute()
352     *
353     */
354    public function getRevs()
355    {
356        if ($this->result_revs === null)
357            throw new StructException('revs are only accessible after executing the search');
358        return $this->result_revs;
359    }
360
361    /**
362     * Execute this search and return the result
363     *
364     * The result is a two dimensional array of Value()s.
365     *
366     * This will always query for the full result (not using offset and limit) and then
367     * return the wanted range, setting the count (@return Value[][]
368     * @see getCount) to the whole result number
369     *
370     */
371    public function execute()
372    {
373        list($sql, $opts) = $this->getSQL();
374
375        /** @var \PDOStatement $res */
376        $res = $this->sqlite->query($sql, $opts);
377        if ($res === false) throw new StructException("SQL execution failed for\n\n$sql");
378
379        $this->result_pids = array();
380        $result = array();
381        $cursor = -1;
382        $pageidAndRevOnly = array_reduce($this->columns, function ($pageidAndRevOnly, Column $col) {
383            return $pageidAndRevOnly && ($col->getTid() == 0);
384        }, true);
385        while ($row = $res->fetch(\PDO::FETCH_ASSOC)) {
386            $cursor++;
387            if ($cursor < $this->range_begin) continue;
388            if ($this->range_end && $cursor >= $this->range_end) continue;
389
390            $C = 0;
391            $resrow = array();
392            $isempty = true;
393            foreach ($this->columns as $col) {
394                $val = $row["C$C"];
395                if ($col->isMulti()) {
396                    $val = explode(self::CONCAT_SEPARATOR, $val);
397                }
398                $value = new Value($col, $val);
399                $isempty &= $this->isEmptyValue($value);
400                $resrow[] = $value;
401                $C++;
402            }
403
404            // skip empty rows
405            if ($isempty && !$pageidAndRevOnly) {
406                $cursor--;
407                continue;
408            }
409
410            $this->result_pids[] = $row['PID'];
411            $this->result_rids[] = $row['rid'];
412            $this->result_revs[] = $row['rev'];
413            $result[] = $resrow;
414        }
415
416        $this->sqlite->res_close($res);
417        $this->count = $cursor + 1;
418        return $result;
419    }
420
421    /**
422     * Transform the set search parameters into a statement
423     *
424     * @return array ($sql, $opts) The SQL and parameters to execute
425     */
426    public function getSQL()
427    {
428        if (!$this->columns) throw new StructException('nocolname');
429
430        $QB = new QueryBuilder();
431
432        // basic tables
433        $first_table = '';
434        foreach ($this->schemas as $schema) {
435            $datatable = 'data_' . $schema->getTable();
436            if ($first_table) {
437                // follow up tables
438                $QB->addLeftJoin($first_table, $datatable, $datatable, "$first_table.pid = $datatable.pid");
439            } else {
440                // first table
441                $QB->addTable($datatable);
442
443                // add conditional page clauses if pid has a value
444                $subAnd = $QB->filters()->whereSubAnd();
445                $subAnd->whereAnd("$datatable.pid = ''");
446                $subOr = $subAnd->whereSubOr();
447                $subOr->whereAnd("GETACCESSLEVEL($datatable.pid) > 0");
448                $subOr->whereAnd("PAGEEXISTS($datatable.pid) = 1");
449                $subOr->whereAnd('(ASSIGNED = 1 OR ASSIGNED IS NULL)');
450
451                // add conditional schema assignment check
452                $QB->addLeftJoin(
453                    $datatable,
454                    'schema_assignments',
455                    '',
456                    "$datatable.pid != ''
457                    AND $datatable.pid = schema_assignments.pid
458                    AND schema_assignments.tbl = '{$schema->getTable()}'"
459                );
460
461                $QB->addSelectColumn($datatable, 'rid');
462                $QB->addSelectColumn($datatable, 'pid', 'PID');
463                $QB->addSelectColumn($datatable, 'rev');
464                $QB->addSelectColumn('schema_assignments', 'assigned', 'ASSIGNED');
465                $QB->addGroupByColumn($datatable, 'pid');
466                $QB->addGroupByColumn($datatable, 'rid');
467
468                $first_table = $datatable;
469            }
470            $QB->filters()->whereAnd("$datatable.latest = 1");
471        }
472
473        // columns to select, handling multis
474        $sep = self::CONCAT_SEPARATOR;
475        $n = 0;
476        foreach ($this->columns as $col) {
477            $CN = 'C' . $n++;
478
479            if ($col->isMulti()) {
480                $datatable = "data_{$col->getTable()}";
481                $multitable = "multi_{$col->getTable()}";
482                $MN = $QB->generateTableAlias('M');
483
484                $QB->addLeftJoin(
485                    $datatable,
486                    $multitable,
487                    $MN,
488                    "$datatable.pid = $MN.pid AND $datatable.rid = $MN.rid AND
489                     $datatable.rev = $MN.rev AND
490                     $MN.colref = {$col->getColref()}"
491                );
492
493                $col->getType()->select($QB, $MN, 'value', $CN);
494                $sel = $QB->getSelectStatement($CN);
495                $QB->addSelectStatement("GROUP_CONCAT($sel, '$sep')", $CN);
496            } else {
497                $col->getType()->select($QB, 'data_' . $col->getTable(), $col->getColName(), $CN);
498                $QB->addGroupByStatement($CN);
499            }
500        }
501
502        // where clauses
503        if (!empty($this->filter)) {
504            $userWHERE = $QB->filters()->where('AND');
505        }
506        foreach ($this->filter as $filter) {
507            /** @var Column $col */
508            list($col, $value, $comp, $op) = $filter;
509
510            $datatable = "data_{$col->getTable()}";
511            $multitable = "multi_{$col->getTable()}";
512
513            /** @var $col Column */
514            if ($col->isMulti()) {
515                $MN = $QB->generateTableAlias('MN');
516
517                $QB->addLeftJoin(
518                    $datatable,
519                    $multitable,
520                    $MN,
521                    "$datatable.pid = $MN.pid AND $datatable.rid = $MN.rid AND
522                     $datatable.rev = $MN.rev AND
523                     $MN.colref = {$col->getColref()}"
524                );
525                $coltbl = $MN;
526                $colnam = 'value';
527            } else {
528                $coltbl = $datatable;
529                $colnam = $col->getColName();
530            }
531
532            $col->getType()->filter($userWHERE, $coltbl, $colnam, $comp, $value, $op); // type based filter
533        }
534
535        // sorting - we always sort by the single val column
536        foreach ($this->sortby as $sort) {
537            list($col, $asc, $nc) = $sort;
538            /** @var $col Column */
539            $colname = $col->getColName(false);
540            if ($nc) $colname .= ' COLLATE NOCASE';
541            $col->getType()->sort($QB, 'data_' . $col->getTable(), $colname, $asc ? 'ASC' : 'DESC');
542        }
543
544        return $QB->getSQL();
545    }
546
547    /**
548     * Returns all the columns that where added to the search
549     *
550     * @return Column[]
551     */
552    public function getColumns()
553    {
554        return $this->columns;
555    }
556
557    /**
558     * All the schemas currently added
559     *
560     * @return Schema[]
561     */
562    public function getSchemas()
563    {
564        return array_values($this->schemas);
565    }
566
567    /**
568     * Checks if the given column is a * wildcard
569     *
570     * If it's a wildcard all matching columns are added to the column list, otherwise
571     * nothing happens
572     *
573     * @param string $colname
574     * @return bool was wildcard?
575     */
576    protected function processWildcard($colname)
577    {
578        list($colname, $table) = $this->resolveColumn($colname);
579        if ($colname !== '*') return false;
580
581        // no table given? assume the first is meant
582        if ($table === null) {
583            $schema_list = array_keys($this->schemas);
584            $table = $schema_list[0];
585        }
586
587        $schema = $this->schemas[$table];
588        if (!$schema) return false;
589        $this->columns = array_merge($this->columns, $schema->getColumns(false));
590        return true;
591    }
592
593    /**
594     * Split a given column name into table and column
595     *
596     * Handles Aliases. Table might be null if none given.
597     *
598     * @param $colname
599     * @return array (colname, table)
600     */
601    protected function resolveColumn($colname)
602    {
603        if (!$this->schemas) throw new StructException('noschemas');
604
605        // resolve the alias or table name
606        @list($table, $colname) = explode('.', $colname, 2);
607        if (!$colname) {
608            $colname = $table;
609            $table = null;
610        }
611        if ($table && isset($this->aliases[$table])) {
612            $table = $this->aliases[$table];
613        }
614
615        if (!$colname) throw new StructException('nocolname');
616
617        return array($colname, $table);
618    }
619
620    /**
621     * Find a column to be used in the search
622     *
623     * @param string $colname may contain an alias
624     * @return bool|Column
625     */
626    public function findColumn($colname, $strict = false)
627    {
628        if (!$this->schemas) throw new StructException('noschemas');
629        $schema_list = array_keys($this->schemas);
630
631        // add "fake" column for special col
632        if ($colname == '%pageid%') {
633            return new PageColumn(0, new Page(), $schema_list[0]);
634        }
635        if ($colname == '%title%') {
636            return new PageColumn(0, new Page(array('usetitles' => true)), $schema_list[0]);
637        }
638        if ($colname == '%lastupdate%') {
639            return new RevisionColumn(0, new DateTime(), $schema_list[0]);
640        }
641        if ($colname == '%lasteditor%') {
642            return new UserColumn(0, new User(), $schema_list[0]);
643        }
644        if ($colname == '%lastsummary%') {
645            return new SummaryColumn(0, new AutoSummary(), $schema_list[0]);
646        }
647        if ($colname == '%rowid%') {
648            return new RowColumn(0, new Decimal(), $schema_list[0]);
649        }
650
651        list($colname, $table) = $this->resolveColumn($colname);
652
653        /*
654         * If table name is given search only that, otherwise if no strict behavior
655         * is requested by the caller, try all assigned schemas for matching the
656         * column name.
657         */
658        if ($table !== null && isset($this->schemas[$table])) {
659            $schemas = array($table => $this->schemas[$table]);
660        } elseif ($table === null || !$strict) {
661            $schemas = $this->schemas;
662        } else {
663            return false;
664        }
665
666        // find it
667        $col = false;
668        foreach ($schemas as $schema) {
669            if (empty($schema)) {
670                continue;
671            }
672            $col = $schema->findColumn($colname);
673            if ($col) break;
674        }
675
676        return $col;
677    }
678
679    /**
680     * Check if the given row is empty or references our own row
681     *
682     * @param Value $value
683     * @return bool
684     */
685    protected function isEmptyValue(Value $value)
686    {
687        if ($value->isEmpty()) return true;
688        if ($value->getColumn()->getTid() == 0) return true;
689        return false;
690    }
691}
692