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#[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 #[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
75pub(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
124pub(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, ¶ms, 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
210fn 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}