sofie_db/search/
texts.rs

1use std::sync::Arc;
2
3use sofie_core::{Cast, ColumnReader, Result, SqlType, SqlValue, Uuid};
4
5use super::SEARCH_COLUMN;
6use crate::{Db, Migrations, Read, Step, query_decoded};
7
8pub const REINDEX_TASK: &str = "sofie.search.reindex";
9pub(super) const PAGE: usize = 500;
10
11#[derive(Clone, Debug, PartialEq, Eq)]
12pub struct SearchIndex {
13    pub(crate) module: &'static str,
14    pub(crate) name: &'static str,
15    pub(crate) owner: Option<&'static str>,
16    pub(crate) hidden: Option<&'static str>,
17    pub(crate) fields: &'static [crate::SearchedField],
18}
19
20/// `text` is the record's search text, recomputed from the fields `Searched` names.
21#[derive(Clone, Debug, PartialEq, Eq)]
22pub struct Document {
23    pub id: Uuid,
24    pub text: String,
25    pub owner: Option<Uuid>,
26}
27
28#[derive(Clone)]
29pub struct SearchTexts(Arc<[SearchIndex]>);
30
31impl SearchTexts {
32    #[must_use]
33    pub fn of(migrations: &[Migrations]) -> Self {
34        Self(
35            migrations
36                .iter()
37                .flat_map(|migrations| migrations.aggregates)
38                .filter_map(crate::AggregateTable::search_index)
39                .collect(),
40        )
41    }
42
43    /// Without a search engine the database matches `search_text` itself, so a record written
44    /// before a field became searched is found by it only once this rewrites its text.
45    #[cfg(not(feature = "search-engine"))]
46    pub async fn reindex(&self, read: Read) -> Result<()> {
47        self.rewrite(&read.db).await
48    }
49
50    #[cfg(feature = "search-engine")]
51    pub async fn reindex(&self, read: Read) -> Result<super::Reindexing> {
52        if let Some(installed) = read.db.search.get() {
53            return Ok(super::Reindexing::every_page(&installed.indexes));
54        }
55        self.rewrite(&read.db).await?;
56        Ok(super::Reindexing::every_page(&[]))
57    }
58
59    async fn rewrite(&self, db: &Db) -> Result<()> {
60        for index in self.0.iter() {
61            let mut after = None;
62            loop {
63                let records = read_records(db, index, Records::After(after)).await?;
64                rewrite_stale(db, index, &records).await?;
65                if records.len() < PAGE {
66                    break;
67                }
68                after = records.last().map(|record| record.document.id);
69            }
70        }
71        Ok(())
72    }
73}
74
75/// Each stored `search_text` that differs from the recomputed text, unless the record changed
76/// since it was read: that commit wrote its own.
77pub(super) async fn rewrite_stale(db: &Db, index: &SearchIndex, records: &[Record]) -> Result<()> {
78    let dialect = db.driver().dialect();
79    let sql = format!(
80        "UPDATE {} SET {} = {} WHERE {} = {} AND {} = {}",
81        dialect.qualified(index.module, index.name),
82        dialect.quote(SEARCH_COLUMN),
83        dialect.placeholder(1),
84        dialect.quote("id"),
85        dialect.placeholder(2),
86        dialect.quote("version"),
87        dialect.placeholder(3),
88    );
89    let steps: Vec<Step> = records
90        .iter()
91        .filter(|record| record.stale)
92        .map(|record| {
93            let params = vec![
94                SqlValue::Text(record.document.text.clone()),
95                SqlValue::Uuid(record.document.id),
96                record.version.clone(),
97            ];
98            Step::statement(sql.clone(), params, None)
99        })
100        .collect();
101    if steps.is_empty() {
102        return Ok(());
103    }
104    db.driver().transaction(steps).await
105}
106
107pub(super) enum Records {
108    #[cfg(feature = "search-engine")]
109    One(Uuid),
110    After(Option<Uuid>),
111}
112
113pub(super) struct Record {
114    pub(super) document: Document,
115    #[cfg_attr(
116        not(feature = "search-engine"),
117        expect(dead_code, reason = "only an engine leaves hidden records out")
118    )]
119    pub(super) hidden: bool,
120    pub(super) version: SqlValue,
121    stale: bool,
122}
123
124/// Deleted records too, so a job can remove them. The text is recomputed from the fields searched
125/// now, so one a field no longer searched (made `#[sensitive]`) still held leaves the engine.
126pub(super) async fn read_records(
127    db: &Db,
128    index: &SearchIndex,
129    records: Records,
130) -> Result<Vec<Record>> {
131    let dialect = db.driver().dialect();
132    let text_columns = super::text_columns(index.fields);
133    let mut columns = vec!["id", "version", SEARCH_COLUMN];
134    let mut types = vec![Some(SqlType::Uuid), Some(SqlType::Int), Some(SqlType::Text)];
135    if let Some(owner) = index.owner {
136        columns.push(owner);
137        types.push(Some(SqlType::Uuid));
138    }
139    if let Some(hidden) = index.hidden {
140        columns.push(hidden);
141        types.push(None);
142    }
143    columns.extend(text_columns.iter().map(|(column, _)| column.as_str()));
144    types.extend(text_columns.iter().map(|_| None));
145    let selected: Vec<String> = columns.iter().map(|column| dialect.quote(column)).collect();
146    let (id, first) = (dialect.quote("id"), dialect.placeholder(1));
147    let (rows, params) = match records {
148        #[cfg(feature = "search-engine")]
149        Records::One(record) => (
150            format!("WHERE {id} = {first}"),
151            vec![SqlValue::Uuid(record)],
152        ),
153        Records::After(Some(after)) => (
154            format!("WHERE {id} > {first} ORDER BY {id} LIMIT {PAGE}"),
155            vec![SqlValue::Uuid(after)],
156        ),
157        Records::After(None) => (format!("ORDER BY {id} LIMIT {PAGE}"), Vec::new()),
158    };
159    let sql = format!(
160        "SELECT {} FROM {} {rows}",
161        selected.join(", "),
162        dialect.qualified(index.module, index.name)
163    );
164    query_decoded(db.driver(), &sql, &params, types, |row| {
165        decode_record(index, &text_columns, row)
166    })
167    .await
168}
169
170fn decode_record(
171    index: &SearchIndex,
172    text_columns: &[(String, SqlType)],
173    row: &mut dyn ColumnReader,
174) -> Result<Record> {
175    let id = Uuid::decode(row)?;
176    let version = SqlValue::Int(i64::decode(row)?);
177    let stored = String::decode(row)?;
178    let owner = match index.owner {
179        Some(_) => Option::<Uuid>::decode(row)?,
180        None => None,
181    };
182    let hidden = match index.hidden {
183        Some(_) => {
184            let hidden = !row.is_null()?;
185            row.skip()?;
186            hidden
187        }
188        None => false,
189    };
190    let values = text_columns
191        .iter()
192        .map(|(_, ty)| searched_value(*ty, row))
193        .collect::<Result<Vec<_>>>()?;
194    let text = if index.fields.is_empty() {
195        stored.clone()
196    } else {
197        super::search_text(index.fields, |column| {
198            let position = text_columns.iter().position(|(name, _)| name == column)?;
199            values.get(position)
200        })
201    };
202    Ok(Record {
203        stale: text != stored,
204        document: Document { id, text, owner },
205        hidden,
206        version,
207    })
208}
209
210/// Only text and ids make words; any other value reads as nothing.
211fn searched_value(ty: SqlType, row: &mut dyn ColumnReader) -> Result<SqlValue> {
212    if row.is_null()? {
213        row.skip()?;
214        return Ok(SqlValue::Null);
215    }
216    match ty {
217        SqlType::Text => Ok(SqlValue::Text(row.text()?.to_owned())),
218        SqlType::Uuid => Ok(SqlValue::Uuid(row.uuid()?)),
219        _ => {
220            row.skip()?;
221            Ok(SqlValue::Null)
222        }
223    }
224}
sofie_db/search/texts.rs

sofie_db/search/
texts.rs
