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