aven_core/metadata/
values.rs

1use std::collections::HashMap;
2
3use anyhow::{Result, bail};
4use serde_json::json;
5use sqlx::{QueryBuilder, Row, Sqlite, SqliteConnection};
6
7use crate::db::{
8    Database, conflict_exists, entity_conflict_exists, entity_field_version, field_version,
9    insert_change, set_entity_field_version, set_field_version,
10};
11use crate::ids::{TaskId, WorkspaceId, now};
12use crate::recurrence::RecurrenceSeriesId;
13use crate::types::MutableEntityType;
14use crate::workspaces::Workspace;
15
16use super::fields::normalize_metadata_key;
17use super::fields::{
18    metadata_field_by_key, require_metadata_field, resolve_or_create_metadata_field,
19};
20use super::validation::validate_metadata_update;
21use super::{MetadataField, ResolvedMetadataValue, TaskMetadataInput, TaskMetadataValue};
22
23impl Database {
24    pub async fn task_metadata(
25        &self,
26        workspace_id: &WorkspaceId,
27        task_id: &TaskId,
28    ) -> Result<Vec<TaskMetadataValue>> {
29        let mut conn = self.acquire_reader().await?;
30        task_metadata_in_workspace(&mut conn, workspace_id, task_id).await
31    }
32}
33
34pub(crate) async fn insert_initial_task_metadata(
35    conn: &mut SqliteConnection,
36    workspace: &Workspace,
37    task_id: &TaskId,
38    inputs: &[TaskMetadataInput],
39    timestamp: &str,
40) -> Result<Vec<ResolvedMetadataValue>> {
41    let values = resolve_metadata_inputs(conn, workspace, inputs).await?;
42    for value in &values {
43        sqlx::query(
44            "INSERT INTO task_metadata(
45                 workspace_id, task_id, field_id, value, created_at, updated_at
46             ) VALUES (?, ?, ?, ?, ?, ?)",
47        )
48        .bind(&workspace.id)
49        .bind(task_id)
50        .bind(&value.field_id)
51        .bind(&value.value)
52        .bind(timestamp)
53        .bind(timestamp)
54        .execute(&mut *conn)
55        .await?;
56    }
57    Ok(values)
58}
59
60pub(crate) async fn set_task_metadata(
61    conn: &mut SqliteConnection,
62    workspace: &Workspace,
63    task_id: &TaskId,
64    input: &TaskMetadataInput,
65) -> Result<bool> {
66    let field = resolve_metadata_input_field(conn, workspace, input).await?;
67    let identity = format!("metadata:{}", field.id);
68    let current = sqlx::query_scalar::<_, String>(
69        "SELECT value FROM task_metadata
70         WHERE workspace_id = ? AND task_id = ? AND field_id = ?",
71    )
72    .bind(&workspace.id)
73    .bind(task_id)
74    .bind(&field.id)
75    .fetch_optional(&mut *conn)
76    .await?;
77    if current.as_deref() == Some(input.value.as_str()) {
78        return Ok(false);
79    }
80    if conflict_exists(conn, &workspace.id, task_id, &identity).await? {
81        bail!("error conflicted-field ref={task_id} field={identity}");
82    }
83    let base = field_version(conn, task_id, &identity).await?;
84    let timestamp = now();
85    sqlx::query(
86        "INSERT INTO task_metadata(
87             workspace_id, task_id, field_id, value, created_at, updated_at
88         ) VALUES (?, ?, ?, ?, ?, ?)
89         ON CONFLICT(workspace_id, task_id, field_id)
90         DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
91    )
92    .bind(&workspace.id)
93    .bind(task_id)
94    .bind(&field.id)
95    .bind(&input.value)
96    .bind(&timestamp)
97    .bind(&timestamp)
98    .execute(&mut *conn)
99    .await?;
100    touch_task(conn, &workspace.id, task_id, &timestamp).await?;
101    let change_id = insert_change(
102        conn,
103        "task",
104        task_id,
105        Some(&identity),
106        "set_task_metadata",
107        json!({
108            "workspace_id": &workspace.id,
109            "workspace_key": &workspace.key,
110            "field_id": &field.id,
111            "key": &field.key,
112            "value": &input.value,
113        }),
114        base.as_deref(),
115    )
116    .await?;
117    set_field_version(conn, task_id, &identity, &change_id).await?;
118    Ok(true)
119}
120
121pub(crate) async fn remove_task_metadata(
122    conn: &mut SqliteConnection,
123    workspace: &Workspace,
124    task_id: &TaskId,
125    input: &str,
126) -> Result<bool> {
127    let key = normalize_metadata_key(input)?;
128    let Some(field) = metadata_field_by_key(conn, &workspace.id, &key).await? else {
129        bail!("error unknown-metadata-field");
130    };
131    let identity = format!("metadata:{}", field.id);
132    if conflict_exists(conn, &workspace.id, task_id, &identity).await? {
133        bail!("error conflicted-field ref={task_id} field={identity}");
134    }
135    let base = field_version(conn, task_id, &identity).await?;
136    let removed_value = sqlx::query_scalar::<_, String>(
137        "SELECT value FROM task_metadata
138         WHERE workspace_id = ? AND task_id = ? AND field_id = ?",
139    )
140    .bind(&workspace.id)
141    .bind(task_id)
142    .bind(&field.id)
143    .fetch_optional(&mut *conn)
144    .await?;
145    let deleted = sqlx::query(
146        "DELETE FROM task_metadata
147         WHERE workspace_id = ? AND task_id = ? AND field_id = ?",
148    )
149    .bind(&workspace.id)
150    .bind(task_id)
151    .bind(&field.id)
152    .execute(&mut *conn)
153    .await?
154    .rows_affected();
155    if deleted == 0 {
156        return Ok(false);
157    }
158    let timestamp = now();
159    touch_task(conn, &workspace.id, task_id, &timestamp).await?;
160    let change_id = insert_change(
161        conn,
162        "task",
163        task_id,
164        Some(&identity),
165        "remove_task_metadata",
166        json!({
167            "workspace_id": &workspace.id,
168            "workspace_key": &workspace.key,
169            "field_id": &field.id,
170            "key": &field.key,
171            "value": removed_value,
172        }),
173        base.as_deref(),
174    )
175    .await?;
176    set_field_version(conn, task_id, &identity, &change_id).await?;
177    Ok(true)
178}
179
180pub(crate) async fn set_recurrence_metadata(
181    conn: &mut SqliteConnection,
182    workspace: &Workspace,
183    series_id: &RecurrenceSeriesId,
184    input: &TaskMetadataInput,
185) -> Result<bool> {
186    let field = resolve_metadata_input_field(conn, workspace, input).await?;
187    let identity = format!("metadata:{}", field.id);
188    let current = sqlx::query_scalar::<_, String>(
189        "SELECT value FROM recurrence_series_metadata
190         WHERE workspace_id = ? AND series_id = ? AND field_id = ?",
191    )
192    .bind(&workspace.id)
193    .bind(series_id)
194    .bind(&field.id)
195    .fetch_optional(&mut *conn)
196    .await?;
197    if current.as_deref() == Some(input.value.as_str()) {
198        return Ok(false);
199    }
200    if entity_conflict_exists(
201        conn,
202        &workspace.id,
203        MutableEntityType::RecurrenceSeries,
204        series_id.as_str(),
205        &identity,
206    )
207    .await?
208    {
209        bail!("error conflicted-recurrence-field field={identity}");
210    }
211    let base = entity_field_version(
212        conn,
213        &workspace.id,
214        MutableEntityType::RecurrenceSeries,
215        series_id.as_str(),
216        &identity,
217    )
218    .await?;
219    let timestamp = now();
220    sqlx::query(
221        "INSERT INTO recurrence_series_metadata(
222             workspace_id, series_id, field_id, value, created_at, updated_at
223         ) VALUES (?, ?, ?, ?, ?, ?)
224         ON CONFLICT(workspace_id, series_id, field_id)
225         DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
226    )
227    .bind(&workspace.id)
228    .bind(series_id)
229    .bind(&field.id)
230    .bind(&input.value)
231    .bind(&timestamp)
232    .bind(&timestamp)
233    .execute(&mut *conn)
234    .await?;
235    sqlx::query("UPDATE recurrence_series SET updated_at = ? WHERE workspace_id = ? AND id = ?")
236        .bind(&timestamp)
237        .bind(&workspace.id)
238        .bind(series_id)
239        .execute(&mut *conn)
240        .await?;
241    let change_id = insert_change(
242        conn,
243        "recurrence_series",
244        series_id.as_str(),
245        Some(&identity),
246        "set_recurrence_metadata",
247        json!({
248            "workspace_id": &workspace.id,
249            "workspace_key": &workspace.key,
250            "field_id": &field.id,
251            "key": &field.key,
252            "value": &input.value,
253        }),
254        base.as_deref(),
255    )
256    .await?;
257    set_entity_field_version(
258        conn,
259        &workspace.id,
260        MutableEntityType::RecurrenceSeries,
261        series_id.as_str(),
262        &identity,
263        &change_id,
264    )
265    .await?;
266    Ok(true)
267}
268
269pub(crate) async fn remove_recurrence_metadata(
270    conn: &mut SqliteConnection,
271    workspace: &Workspace,
272    series_id: &RecurrenceSeriesId,
273    key: &str,
274) -> Result<bool> {
275    let Some(field) =
276        metadata_field_by_key(conn, &workspace.id, &normalize_metadata_key(key)?).await?
277    else {
278        return Ok(false);
279    };
280    let identity = format!("metadata:{}", field.id);
281    if entity_conflict_exists(
282        conn,
283        &workspace.id,
284        MutableEntityType::RecurrenceSeries,
285        series_id.as_str(),
286        &identity,
287    )
288    .await?
289    {
290        bail!("error conflicted-recurrence-field field={identity}");
291    }
292    let removed = sqlx::query(
293        "DELETE FROM recurrence_series_metadata
294         WHERE workspace_id = ? AND series_id = ? AND field_id = ?",
295    )
296    .bind(&workspace.id)
297    .bind(series_id)
298    .bind(&field.id)
299    .execute(&mut *conn)
300    .await?
301    .rows_affected();
302    if removed == 0 {
303        return Ok(false);
304    }
305    let base = entity_field_version(
306        conn,
307        &workspace.id,
308        MutableEntityType::RecurrenceSeries,
309        series_id.as_str(),
310        &identity,
311    )
312    .await?;
313    let timestamp = now();
314    sqlx::query("UPDATE recurrence_series SET updated_at = ? WHERE workspace_id = ? AND id = ?")
315        .bind(&timestamp)
316        .bind(&workspace.id)
317        .bind(series_id)
318        .execute(&mut *conn)
319        .await?;
320    let change_id = insert_change(
321        conn,
322        "recurrence_series",
323        series_id.as_str(),
324        Some(&identity),
325        "remove_recurrence_metadata",
326        json!({
327            "workspace_id": &workspace.id,
328            "workspace_key": &workspace.key,
329            "field_id": &field.id,
330            "key": &field.key,
331        }),
332        base.as_deref(),
333    )
334    .await?;
335    set_entity_field_version(
336        conn,
337        &workspace.id,
338        MutableEntityType::RecurrenceSeries,
339        series_id.as_str(),
340        &identity,
341        &change_id,
342    )
343    .await?;
344    Ok(true)
345}
346async fn resolve_metadata_input_field(
347    conn: &mut SqliteConnection,
348    workspace: &Workspace,
349    input: &TaskMetadataInput,
350) -> Result<MetadataField> {
351    match &input.expected_field_id {
352        Some(id) => require_metadata_field(conn, &workspace.id, id, &input.key).await,
353        None => resolve_or_create_metadata_field(conn, workspace, &input.key).await,
354    }
355}
356
357pub(crate) async fn resolve_metadata_inputs(
358    conn: &mut SqliteConnection,
359    workspace: &Workspace,
360    inputs: &[TaskMetadataInput],
361) -> Result<Vec<ResolvedMetadataValue>> {
362    validate_metadata_update(inputs, &[])?;
363    let mut values = Vec::with_capacity(inputs.len());
364    for input in inputs {
365        let field = resolve_metadata_input_field(conn, workspace, input).await?;
366        values.push(ResolvedMetadataValue {
367            field_id: field.id,
368            key: field.key,
369            value: input.value.clone(),
370        });
371    }
372    Ok(values)
373}
374async fn touch_task(
375    conn: &mut SqliteConnection,
376    workspace_id: &WorkspaceId,
377    task_id: &TaskId,
378    timestamp: &str,
379) -> Result<()> {
380    let affected = sqlx::query("UPDATE tasks SET updated_at = ? WHERE workspace_id = ? AND id = ?")
381        .bind(timestamp)
382        .bind(workspace_id)
383        .bind(task_id)
384        .execute(&mut *conn)
385        .await?
386        .rows_affected();
387    if affected != 1 {
388        bail!("error task-not-found task_id={task_id}");
389    }
390    Ok(())
391}
392
393pub(crate) async fn task_metadata_in_workspace(
394    conn: &mut SqliteConnection,
395    workspace_id: &WorkspaceId,
396    task_id: &TaskId,
397) -> Result<Vec<TaskMetadataValue>> {
398    let rows = sqlx::query(
399        "SELECT m.field_id, f.key, m.value
400         FROM task_metadata m
401         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
402         WHERE m.workspace_id = ? AND m.task_id = ?
403         ORDER BY f.key",
404    )
405    .bind(workspace_id)
406    .bind(task_id)
407    .fetch_all(&mut *conn)
408    .await?;
409    Ok(rows
410        .into_iter()
411        .map(|row| TaskMetadataValue {
412            field_id: row.get("field_id"),
413            key: row.get("key"),
414            value: row.get("value"),
415        })
416        .collect())
417}
418
419pub(crate) async fn metadata_by_task_ids(
420    conn: &mut SqliteConnection,
421    workspace_id: &WorkspaceId,
422    task_ids: &[TaskId],
423) -> Result<HashMap<TaskId, Vec<TaskMetadataValue>>> {
424    let mut values = HashMap::new();
425    for chunk in task_ids.chunks(900) {
426        let mut query = QueryBuilder::<Sqlite>::new(
427            "SELECT m.task_id, m.field_id, f.key, m.value
428             FROM task_metadata m
429             JOIN metadata_fields f
430               ON f.workspace_id = m.workspace_id AND f.id = m.field_id
431             WHERE m.workspace_id = ",
432        );
433        query.push_bind(workspace_id);
434        query.push(" AND m.task_id IN (");
435        {
436            let mut separated = query.separated(", ");
437            for task_id in chunk {
438                separated.push_bind(task_id);
439            }
440        }
441        query.push(") ORDER BY m.task_id, f.key");
442        for row in query.build().fetch_all(&mut *conn).await? {
443            let task_id: TaskId = row.get("task_id");
444            values
445                .entry(task_id)
446                .or_insert_with(Vec::new)
447                .push(TaskMetadataValue {
448                    field_id: row.get("field_id"),
449                    key: row.get("key"),
450                    value: row.get("value"),
451                });
452        }
453    }
454    Ok(values)
455}
aven_core/metadata/values.rs

aven_core/metadata/
values.rs
