aven_core/
metadata.rs

1use std::collections::{BTreeSet, HashMap};
2
3use anyhow::{Result, bail};
4use serde::{Deserialize, Serialize};
5use serde_json::json;
6use sqlx::{QueryBuilder, Row, Sqlite, SqliteConnection};
7
8use crate::db::{
9    Database, begin_immediate, conflict_exists, entity_conflict_exists, entity_field_version,
10    field_version, insert_change, set_entity_field_version, set_field_version,
11};
12use crate::ids::{MetadataFieldId, TaskId, WorkspaceId, now};
13use crate::recurrence::RecurrenceSeriesId;
14use crate::types::MutableEntityType;
15use crate::workspaces::Workspace;
16
17pub(crate) const MAX_METADATA_VALUES: usize = 128;
18pub(crate) const MAX_METADATA_VALUE_BYTES: usize = 4 * 1024;
19pub(crate) const MAX_METADATA_TOTAL_BYTES: usize = 32 * 1024;
20
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct MetadataField {
23    pub id: MetadataFieldId,
24    pub workspace_id: WorkspaceId,
25    pub key: String,
26    pub created_at: String,
27    pub updated_at: String,
28}
29
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct TaskMetadataValue {
32    pub field_id: MetadataFieldId,
33    pub key: String,
34    pub value: String,
35}
36
37#[derive(Debug, Clone, PartialEq, Eq)]
38pub struct TaskMetadataInput {
39    pub key: String,
40    pub value: String,
41}
42
43#[derive(Debug, Clone, PartialEq, Eq)]
44pub struct MetadataFieldUsage {
45    pub field: MetadataField,
46    pub task_count: usize,
47    pub series_count: usize,
48}
49
50#[derive(Debug, Clone, Serialize)]
51pub(crate) struct ResolvedMetadataValue {
52    pub(crate) field_id: MetadataFieldId,
53    pub(crate) key: String,
54    pub(crate) value: String,
55}
56
57#[derive(Serialize, Deserialize)]
58struct MetadataConflictValue {
59    present: bool,
60    value: String,
61}
62
63pub(crate) fn encode_metadata_conflict_value(value: Option<&str>) -> Result<String> {
64    Ok(serde_json::to_string(&MetadataConflictValue {
65        present: value.is_some(),
66        value: value.unwrap_or_default().to_string(),
67    })?)
68}
69
70pub(crate) fn decode_metadata_conflict_value(value: &str) -> Result<Option<String>> {
71    let value: MetadataConflictValue = serde_json::from_str(value)?;
72    Ok(value.present.then_some(value.value))
73}
74
75impl Database {
76    pub async fn list_metadata_fields(
77        &self,
78        workspace_id: &WorkspaceId,
79    ) -> Result<Vec<MetadataFieldUsage>> {
80        let mut conn = self.acquire_reader().await?;
81        list_metadata_fields_in_workspace(&mut conn, workspace_id).await
82    }
83
84    pub async fn find_metadata_field(
85        &self,
86        workspace_id: &WorkspaceId,
87        key: &str,
88    ) -> Result<Option<MetadataField>> {
89        let mut conn = self.acquire_reader().await?;
90        find_metadata_field_in_workspace(&mut conn, workspace_id, key).await
91    }
92
93    pub async fn task_metadata(
94        &self,
95        workspace_id: &WorkspaceId,
96        task_id: &TaskId,
97    ) -> Result<Vec<TaskMetadataValue>> {
98        let mut conn = self.acquire_reader().await?;
99        task_metadata_in_workspace(&mut conn, workspace_id, task_id).await
100    }
101
102    pub async fn rename_metadata_field(
103        &self,
104        workspace: &Workspace,
105        key: &str,
106        new_key: &str,
107    ) -> Result<MetadataField> {
108        let mut conn = self.acquire_writer().await?;
109        let mut tx = begin_immediate(&mut conn).await?;
110        let field = rename_metadata_field(&mut tx, workspace, key, new_key).await?;
111        tx.commit().await?;
112        Ok(field)
113    }
114}
115
116pub fn normalize_metadata_key(input: &str) -> Result<String> {
117    let key = input.trim().to_ascii_lowercase();
118    let mut bytes = key.bytes();
119    let Some(first) = bytes.next() else {
120        bail!("error invalid-metadata-key");
121    };
122    if !first.is_ascii_lowercase()
123        || key.len() > 64
124        || !bytes.all(|byte| {
125            byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'.' | b'_' | b'-')
126        })
127        || key.starts_with("aven.")
128    {
129        bail!("error invalid-metadata-key");
130    }
131    Ok(key)
132}
133
134pub(crate) async fn find_metadata_field_in_workspace(
135    conn: &mut SqliteConnection,
136    workspace_id: &WorkspaceId,
137    input: &str,
138) -> Result<Option<MetadataField>> {
139    let key = normalize_metadata_key(input)?;
140    metadata_field_by_key(conn, workspace_id, &key).await
141}
142
143pub(crate) async fn metadata_field_by_key(
144    conn: &mut SqliteConnection,
145    workspace_id: &WorkspaceId,
146    key: &str,
147) -> Result<Option<MetadataField>> {
148    let row = sqlx::query(
149        "SELECT id, workspace_id, key, created_at, updated_at
150         FROM metadata_fields WHERE workspace_id = ? AND key = ?",
151    )
152    .bind(workspace_id)
153    .bind(key)
154    .fetch_optional(&mut *conn)
155    .await?;
156    Ok(row.map(metadata_field_from_row))
157}
158
159pub(crate) async fn metadata_field_by_id(
160    conn: &mut SqliteConnection,
161    workspace_id: &WorkspaceId,
162    field_id: &MetadataFieldId,
163) -> Result<Option<MetadataField>> {
164    let row = sqlx::query(
165        "SELECT id, workspace_id, key, created_at, updated_at
166         FROM metadata_fields WHERE workspace_id = ? AND id = ?",
167    )
168    .bind(workspace_id)
169    .bind(field_id)
170    .fetch_optional(&mut *conn)
171    .await?;
172    Ok(row.map(metadata_field_from_row))
173}
174
175pub(crate) async fn resolve_or_create_metadata_field(
176    conn: &mut SqliteConnection,
177    workspace: &Workspace,
178    input: &str,
179) -> Result<MetadataField> {
180    let key = normalize_metadata_key(input)?;
181    if let Some(field) = metadata_field_by_key(conn, &workspace.id, &key).await? {
182        return Ok(field);
183    }
184
185    let id = MetadataFieldId::new();
186    let timestamp = now();
187    sqlx::query(
188        "INSERT INTO metadata_fields(id, workspace_id, key, created_at, updated_at)
189         VALUES (?, ?, ?, ?, ?)",
190    )
191    .bind(&id)
192    .bind(&workspace.id)
193    .bind(&key)
194    .bind(&timestamp)
195    .bind(&timestamp)
196    .execute(&mut *conn)
197    .await?;
198    let change_id = insert_change(
199        conn,
200        "metadata_field",
201        id.as_str(),
202        None,
203        "create_metadata_field",
204        json!({
205            "workspace_id": &workspace.id,
206            "workspace_key": &workspace.key,
207            "key": &key,
208            "created_at": &timestamp,
209        }),
210        None,
211    )
212    .await?;
213    set_entity_field_version(
214        conn,
215        &workspace.id,
216        MutableEntityType::MetadataField,
217        id.as_str(),
218        "key",
219        &change_id,
220    )
221    .await?;
222    Ok(MetadataField {
223        id,
224        workspace_id: workspace.id.clone(),
225        key,
226        created_at: timestamp.clone(),
227        updated_at: timestamp,
228    })
229}
230
231pub(crate) async fn rename_metadata_field(
232    conn: &mut SqliteConnection,
233    workspace: &Workspace,
234    input: &str,
235    new_input: &str,
236) -> Result<MetadataField> {
237    let key = normalize_metadata_key(input)?;
238    let new_key = normalize_metadata_key(new_input)?;
239    let mut field = metadata_field_by_key(conn, &workspace.id, &key)
240        .await?
241        .ok_or_else(|| anyhow::anyhow!("error unknown-metadata-field"))?;
242    let resolving = entity_conflict_exists(
243        conn,
244        &workspace.id,
245        MutableEntityType::MetadataField,
246        field.id.as_str(),
247        "key",
248    )
249    .await?;
250    if key == new_key && !resolving {
251        return Ok(field);
252    }
253    if key != new_key
254        && metadata_field_by_key(conn, &workspace.id, &new_key)
255            .await?
256            .is_some()
257    {
258        bail!("error metadata-field-exists");
259    }
260    let base = if resolving {
261        None
262    } else {
263        entity_field_version(
264            conn,
265            &workspace.id,
266            MutableEntityType::MetadataField,
267            field.id.as_str(),
268            "key",
269        )
270        .await?
271    };
272    let timestamp = now();
273    sqlx::query(
274        "UPDATE metadata_fields SET key = ?, updated_at = ?
275         WHERE workspace_id = ? AND id = ?",
276    )
277    .bind(&new_key)
278    .bind(&timestamp)
279    .bind(&workspace.id)
280    .bind(&field.id)
281    .execute(&mut *conn)
282    .await?;
283    let change_id = insert_change(
284        conn,
285        "metadata_field",
286        field.id.as_str(),
287        Some("key"),
288        "set_metadata_field",
289        json!({
290            "workspace_id": &workspace.id,
291            "workspace_key": &workspace.key,
292            "key": &new_key,
293            "conflict_resolution": resolving,
294        }),
295        base.as_deref(),
296    )
297    .await?;
298    set_entity_field_version(
299        conn,
300        &workspace.id,
301        MutableEntityType::MetadataField,
302        field.id.as_str(),
303        "key",
304        &change_id,
305    )
306    .await?;
307    if resolving {
308        sqlx::query(
309            "UPDATE conflicts SET resolved = 1
310             WHERE workspace_id = ? AND entity_type = 'metadata_field'
311               AND entity_id = ? AND field = 'key' AND resolved = 0",
312        )
313        .bind(&workspace.id)
314        .bind(&field.id)
315        .execute(&mut *conn)
316        .await?;
317    }
318    field.key = new_key;
319    field.updated_at = timestamp;
320    Ok(field)
321}
322
323pub(crate) async fn insert_initial_task_metadata(
324    conn: &mut SqliteConnection,
325    workspace: &Workspace,
326    task_id: &TaskId,
327    inputs: &[TaskMetadataInput],
328    timestamp: &str,
329) -> Result<Vec<ResolvedMetadataValue>> {
330    let values = resolve_metadata_inputs(conn, workspace, inputs).await?;
331    for value in &values {
332        sqlx::query(
333            "INSERT INTO task_metadata(
334                 workspace_id, task_id, field_id, value, created_at, updated_at
335             ) VALUES (?, ?, ?, ?, ?, ?)",
336        )
337        .bind(&workspace.id)
338        .bind(task_id)
339        .bind(&value.field_id)
340        .bind(&value.value)
341        .bind(timestamp)
342        .bind(timestamp)
343        .execute(&mut *conn)
344        .await?;
345    }
346    Ok(values)
347}
348
349pub(crate) async fn set_task_metadata(
350    conn: &mut SqliteConnection,
351    workspace: &Workspace,
352    task_id: &TaskId,
353    input: &TaskMetadataInput,
354) -> Result<bool> {
355    let field = resolve_or_create_metadata_field(conn, workspace, &input.key).await?;
356    let identity = format!("metadata:{}", field.id);
357    let current = sqlx::query_scalar::<_, String>(
358        "SELECT value FROM task_metadata
359         WHERE workspace_id = ? AND task_id = ? AND field_id = ?",
360    )
361    .bind(&workspace.id)
362    .bind(task_id)
363    .bind(&field.id)
364    .fetch_optional(&mut *conn)
365    .await?;
366    if current.as_deref() == Some(input.value.as_str()) {
367        return Ok(false);
368    }
369    if conflict_exists(conn, &workspace.id, task_id, &identity).await? {
370        bail!("error conflicted-field ref={task_id} field={identity}");
371    }
372    let base = field_version(conn, task_id, &identity).await?;
373    let timestamp = now();
374    sqlx::query(
375        "INSERT INTO task_metadata(
376             workspace_id, task_id, field_id, value, created_at, updated_at
377         ) VALUES (?, ?, ?, ?, ?, ?)
378         ON CONFLICT(workspace_id, task_id, field_id)
379         DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
380    )
381    .bind(&workspace.id)
382    .bind(task_id)
383    .bind(&field.id)
384    .bind(&input.value)
385    .bind(&timestamp)
386    .bind(&timestamp)
387    .execute(&mut *conn)
388    .await?;
389    touch_task(conn, &workspace.id, task_id, &timestamp).await?;
390    let change_id = insert_change(
391        conn,
392        "task",
393        task_id,
394        Some(&identity),
395        "set_task_metadata",
396        json!({
397            "workspace_id": &workspace.id,
398            "workspace_key": &workspace.key,
399            "field_id": &field.id,
400            "key": &field.key,
401            "value": &input.value,
402        }),
403        base.as_deref(),
404    )
405    .await?;
406    set_field_version(conn, task_id, &identity, &change_id).await?;
407    Ok(true)
408}
409
410pub(crate) async fn remove_task_metadata(
411    conn: &mut SqliteConnection,
412    workspace: &Workspace,
413    task_id: &TaskId,
414    input: &str,
415) -> Result<bool> {
416    let key = normalize_metadata_key(input)?;
417    let Some(field) = metadata_field_by_key(conn, &workspace.id, &key).await? else {
418        bail!("error unknown-metadata-field");
419    };
420    let identity = format!("metadata:{}", field.id);
421    if conflict_exists(conn, &workspace.id, task_id, &identity).await? {
422        bail!("error conflicted-field ref={task_id} field={identity}");
423    }
424    let base = field_version(conn, task_id, &identity).await?;
425    let removed_value = sqlx::query_scalar::<_, String>(
426        "SELECT value FROM task_metadata
427         WHERE workspace_id = ? AND task_id = ? AND field_id = ?",
428    )
429    .bind(&workspace.id)
430    .bind(task_id)
431    .bind(&field.id)
432    .fetch_optional(&mut *conn)
433    .await?;
434    let deleted = sqlx::query(
435        "DELETE FROM task_metadata
436         WHERE workspace_id = ? AND task_id = ? AND field_id = ?",
437    )
438    .bind(&workspace.id)
439    .bind(task_id)
440    .bind(&field.id)
441    .execute(&mut *conn)
442    .await?
443    .rows_affected();
444    if deleted == 0 {
445        return Ok(false);
446    }
447    let timestamp = now();
448    touch_task(conn, &workspace.id, task_id, &timestamp).await?;
449    let change_id = insert_change(
450        conn,
451        "task",
452        task_id,
453        Some(&identity),
454        "remove_task_metadata",
455        json!({
456            "workspace_id": &workspace.id,
457            "workspace_key": &workspace.key,
458            "field_id": &field.id,
459            "key": &field.key,
460            "value": removed_value,
461        }),
462        base.as_deref(),
463    )
464    .await?;
465    set_field_version(conn, task_id, &identity, &change_id).await?;
466    Ok(true)
467}
468
469pub(crate) async fn set_recurrence_metadata(
470    conn: &mut SqliteConnection,
471    workspace: &Workspace,
472    series_id: &RecurrenceSeriesId,
473    input: &TaskMetadataInput,
474) -> Result<bool> {
475    let field = resolve_or_create_metadata_field(conn, workspace, &input.key).await?;
476    let identity = format!("metadata:{}", field.id);
477    let current = sqlx::query_scalar::<_, String>(
478        "SELECT value FROM recurrence_series_metadata
479         WHERE workspace_id = ? AND series_id = ? AND field_id = ?",
480    )
481    .bind(&workspace.id)
482    .bind(series_id)
483    .bind(&field.id)
484    .fetch_optional(&mut *conn)
485    .await?;
486    if current.as_deref() == Some(input.value.as_str()) {
487        return Ok(false);
488    }
489    if entity_conflict_exists(
490        conn,
491        &workspace.id,
492        MutableEntityType::RecurrenceSeries,
493        series_id.as_str(),
494        &identity,
495    )
496    .await?
497    {
498        bail!("error conflicted-recurrence-field field={identity}");
499    }
500    let base = entity_field_version(
501        conn,
502        &workspace.id,
503        MutableEntityType::RecurrenceSeries,
504        series_id.as_str(),
505        &identity,
506    )
507    .await?;
508    let timestamp = now();
509    sqlx::query(
510        "INSERT INTO recurrence_series_metadata(
511             workspace_id, series_id, field_id, value, created_at, updated_at
512         ) VALUES (?, ?, ?, ?, ?, ?)
513         ON CONFLICT(workspace_id, series_id, field_id)
514         DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
515    )
516    .bind(&workspace.id)
517    .bind(series_id)
518    .bind(&field.id)
519    .bind(&input.value)
520    .bind(&timestamp)
521    .bind(&timestamp)
522    .execute(&mut *conn)
523    .await?;
524    sqlx::query("UPDATE recurrence_series SET updated_at = ? WHERE workspace_id = ? AND id = ?")
525        .bind(&timestamp)
526        .bind(&workspace.id)
527        .bind(series_id)
528        .execute(&mut *conn)
529        .await?;
530    let change_id = insert_change(
531        conn,
532        "recurrence_series",
533        series_id.as_str(),
534        Some(&identity),
535        "set_recurrence_metadata",
536        json!({
537            "workspace_id": &workspace.id,
538            "workspace_key": &workspace.key,
539            "field_id": &field.id,
540            "key": &field.key,
541            "value": &input.value,
542        }),
543        base.as_deref(),
544    )
545    .await?;
546    set_entity_field_version(
547        conn,
548        &workspace.id,
549        MutableEntityType::RecurrenceSeries,
550        series_id.as_str(),
551        &identity,
552        &change_id,
553    )
554    .await?;
555    Ok(true)
556}
557
558pub(crate) async fn remove_recurrence_metadata(
559    conn: &mut SqliteConnection,
560    workspace: &Workspace,
561    series_id: &RecurrenceSeriesId,
562    key: &str,
563) -> Result<bool> {
564    let Some(field) =
565        metadata_field_by_key(conn, &workspace.id, &normalize_metadata_key(key)?).await?
566    else {
567        return Ok(false);
568    };
569    let identity = format!("metadata:{}", field.id);
570    if entity_conflict_exists(
571        conn,
572        &workspace.id,
573        MutableEntityType::RecurrenceSeries,
574        series_id.as_str(),
575        &identity,
576    )
577    .await?
578    {
579        bail!("error conflicted-recurrence-field field={identity}");
580    }
581    let removed = sqlx::query(
582        "DELETE FROM recurrence_series_metadata
583         WHERE workspace_id = ? AND series_id = ? AND field_id = ?",
584    )
585    .bind(&workspace.id)
586    .bind(series_id)
587    .bind(&field.id)
588    .execute(&mut *conn)
589    .await?
590    .rows_affected();
591    if removed == 0 {
592        return Ok(false);
593    }
594    let base = entity_field_version(
595        conn,
596        &workspace.id,
597        MutableEntityType::RecurrenceSeries,
598        series_id.as_str(),
599        &identity,
600    )
601    .await?;
602    let timestamp = now();
603    sqlx::query("UPDATE recurrence_series SET updated_at = ? WHERE workspace_id = ? AND id = ?")
604        .bind(&timestamp)
605        .bind(&workspace.id)
606        .bind(series_id)
607        .execute(&mut *conn)
608        .await?;
609    let change_id = insert_change(
610        conn,
611        "recurrence_series",
612        series_id.as_str(),
613        Some(&identity),
614        "remove_recurrence_metadata",
615        json!({
616            "workspace_id": &workspace.id,
617            "workspace_key": &workspace.key,
618            "field_id": &field.id,
619            "key": &field.key,
620        }),
621        base.as_deref(),
622    )
623    .await?;
624    set_entity_field_version(
625        conn,
626        &workspace.id,
627        MutableEntityType::RecurrenceSeries,
628        series_id.as_str(),
629        &identity,
630        &change_id,
631    )
632    .await?;
633    Ok(true)
634}
635
636pub(crate) fn validate_metadata_update(set: &[TaskMetadataInput], remove: &[String]) -> Result<()> {
637    if set.len() > MAX_METADATA_VALUES {
638        bail!("error too-many-metadata-values limit={MAX_METADATA_VALUES}");
639    }
640    let mut keys = BTreeSet::new();
641    let mut total_bytes = 0usize;
642    for input in set {
643        let key = normalize_metadata_key(&input.key)?;
644        if !keys.insert(key) {
645            bail!("error duplicate-metadata-key");
646        }
647        if input.value.len() > MAX_METADATA_VALUE_BYTES {
648            bail!("error metadata-value-too-large limit={MAX_METADATA_VALUE_BYTES}");
649        }
650        total_bytes = total_bytes.saturating_add(input.value.len());
651    }
652    if total_bytes > MAX_METADATA_TOTAL_BYTES {
653        bail!("error metadata-values-too-large limit={MAX_METADATA_TOTAL_BYTES}");
654    }
655    for input in remove {
656        let key = normalize_metadata_key(input)?;
657        if !keys.insert(key) {
658            bail!("error overlapping-metadata-key");
659        }
660    }
661    Ok(())
662}
663
664fn validate_metadata_result_limits(
665    existing: impl IntoIterator<Item = (String, String)>,
666    set: &[TaskMetadataInput],
667    remove: &[String],
668) -> Result<()> {
669    let mut values = existing.into_iter().collect::<HashMap<_, _>>();
670    for key in remove {
671        values.remove(&normalize_metadata_key(key)?);
672    }
673    for input in set {
674        values.insert(normalize_metadata_key(&input.key)?, input.value.clone());
675    }
676    if values.len() > MAX_METADATA_VALUES {
677        bail!("error too-many-metadata-values limit={MAX_METADATA_VALUES}");
678    }
679    let total_bytes = values.values().map(String::len).sum::<usize>();
680    if total_bytes > MAX_METADATA_TOTAL_BYTES {
681        bail!("error metadata-values-too-large limit={MAX_METADATA_TOTAL_BYTES}");
682    }
683    Ok(())
684}
685
686pub(crate) async fn validate_recurrence_metadata_result(
687    conn: &mut SqliteConnection,
688    workspace_id: &WorkspaceId,
689    series_id: &RecurrenceSeriesId,
690    set: &[TaskMetadataInput],
691    remove: &[String],
692) -> Result<()> {
693    validate_metadata_update(set, remove)?;
694    let rows = sqlx::query(
695        "SELECT f.key, m.value FROM recurrence_series_metadata m
696         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
697         WHERE m.workspace_id = ? AND m.series_id = ?",
698    )
699    .bind(workspace_id)
700    .bind(series_id)
701    .fetch_all(&mut *conn)
702    .await?;
703    validate_metadata_result_limits(
704        rows.into_iter()
705            .map(|row| (row.get::<String, _>("key"), row.get::<String, _>("value"))),
706        set,
707        remove,
708    )
709}
710
711pub(crate) async fn validate_task_metadata_result(
712    conn: &mut SqliteConnection,
713    workspace_id: &WorkspaceId,
714    task_id: &TaskId,
715    set: &[TaskMetadataInput],
716    remove: &[String],
717) -> Result<()> {
718    validate_metadata_update(set, remove)?;
719    let rows = sqlx::query(
720        "SELECT f.key, m.value FROM task_metadata m
721         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
722         WHERE m.workspace_id = ? AND m.task_id = ?",
723    )
724    .bind(workspace_id)
725    .bind(task_id)
726    .fetch_all(&mut *conn)
727    .await?;
728    validate_metadata_result_limits(
729        rows.into_iter()
730            .map(|row| (row.get::<String, _>("key"), row.get::<String, _>("value"))),
731        set,
732        remove,
733    )
734}
735
736pub(crate) async fn resolve_metadata_inputs(
737    conn: &mut SqliteConnection,
738    workspace: &Workspace,
739    inputs: &[TaskMetadataInput],
740) -> Result<Vec<ResolvedMetadataValue>> {
741    validate_metadata_update(inputs, &[])?;
742    let mut values = Vec::with_capacity(inputs.len());
743    for input in inputs {
744        let field = resolve_or_create_metadata_field(conn, workspace, &input.key).await?;
745        values.push(ResolvedMetadataValue {
746            field_id: field.id,
747            key: field.key,
748            value: input.value.clone(),
749        });
750    }
751    Ok(values)
752}
753
754async fn touch_task(
755    conn: &mut SqliteConnection,
756    workspace_id: &WorkspaceId,
757    task_id: &TaskId,
758    timestamp: &str,
759) -> Result<()> {
760    let affected = sqlx::query("UPDATE tasks SET updated_at = ? WHERE workspace_id = ? AND id = ?")
761        .bind(timestamp)
762        .bind(workspace_id)
763        .bind(task_id)
764        .execute(&mut *conn)
765        .await?
766        .rows_affected();
767    if affected != 1 {
768        bail!("error task-not-found task_id={task_id}");
769    }
770    Ok(())
771}
772
773pub(crate) async fn task_metadata_in_workspace(
774    conn: &mut SqliteConnection,
775    workspace_id: &WorkspaceId,
776    task_id: &TaskId,
777) -> Result<Vec<TaskMetadataValue>> {
778    let rows = sqlx::query(
779        "SELECT m.field_id, f.key, m.value
780         FROM task_metadata m
781         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
782         WHERE m.workspace_id = ? AND m.task_id = ?
783         ORDER BY f.key",
784    )
785    .bind(workspace_id)
786    .bind(task_id)
787    .fetch_all(&mut *conn)
788    .await?;
789    Ok(rows
790        .into_iter()
791        .map(|row| TaskMetadataValue {
792            field_id: row.get("field_id"),
793            key: row.get("key"),
794            value: row.get("value"),
795        })
796        .collect())
797}
798
799pub(crate) async fn metadata_by_task_ids(
800    conn: &mut SqliteConnection,
801    workspace_id: &WorkspaceId,
802    task_ids: &[TaskId],
803) -> Result<HashMap<TaskId, Vec<TaskMetadataValue>>> {
804    let mut values = HashMap::new();
805    for chunk in task_ids.chunks(900) {
806        let mut query = QueryBuilder::<Sqlite>::new(
807            "SELECT m.task_id, m.field_id, f.key, m.value
808             FROM task_metadata m
809             JOIN metadata_fields f
810               ON f.workspace_id = m.workspace_id AND f.id = m.field_id
811             WHERE m.workspace_id = ",
812        );
813        query.push_bind(workspace_id);
814        query.push(" AND m.task_id IN (");
815        {
816            let mut separated = query.separated(", ");
817            for task_id in chunk {
818                separated.push_bind(task_id);
819            }
820        }
821        query.push(") ORDER BY m.task_id, f.key");
822        for row in query.build().fetch_all(&mut *conn).await? {
823            let task_id: TaskId = row.get("task_id");
824            values
825                .entry(task_id)
826                .or_insert_with(Vec::new)
827                .push(TaskMetadataValue {
828                    field_id: row.get("field_id"),
829                    key: row.get("key"),
830                    value: row.get("value"),
831                });
832        }
833    }
834    Ok(values)
835}
836
837async fn list_metadata_fields_in_workspace(
838    conn: &mut SqliteConnection,
839    workspace_id: &WorkspaceId,
840) -> Result<Vec<MetadataFieldUsage>> {
841    let rows = sqlx::query(
842        "SELECT f.id, f.workspace_id, f.key, f.created_at, f.updated_at,
843                (SELECT count(*) FROM task_metadata m
844                 WHERE m.workspace_id = f.workspace_id AND m.field_id = f.id) AS task_count,
845                (SELECT count(*) FROM recurrence_series_metadata m
846                 WHERE m.workspace_id = f.workspace_id AND m.field_id = f.id) AS series_count
847         FROM metadata_fields f
848         WHERE f.workspace_id = ?
849         ORDER BY f.key",
850    )
851    .bind(workspace_id)
852    .fetch_all(&mut *conn)
853    .await?;
854    Ok(rows
855        .into_iter()
856        .map(|row| MetadataFieldUsage {
857            field: MetadataField {
858                id: row.get("id"),
859                workspace_id: row.get("workspace_id"),
860                key: row.get("key"),
861                created_at: row.get("created_at"),
862                updated_at: row.get("updated_at"),
863            },
864            task_count: usize::try_from(row.get::<i64, _>("task_count")).unwrap_or(usize::MAX),
865            series_count: usize::try_from(row.get::<i64, _>("series_count")).unwrap_or(usize::MAX),
866        })
867        .collect())
868}
869
870fn metadata_field_from_row(row: sqlx::sqlite::SqliteRow) -> MetadataField {
871    MetadataField {
872        id: row.get("id"),
873        workspace_id: row.get("workspace_id"),
874        key: row.get("key"),
875        created_at: row.get("created_at"),
876        updated_at: row.get("updated_at"),
877    }
878}
879
880#[cfg(test)]
881mod tests {
882    use super::*;
883    use crate::db::begin_immediate;
884    use crate::test_support::{ensure_default_workspace, test_conn};
885
886    #[test]
887    fn metadata_keys_have_stable_normalization() {
888        assert_eq!(normalize_metadata_key(" Max_Turns ").unwrap(), "max_turns");
889        assert_eq!(normalize_metadata_key("source.id").unwrap(), "source.id");
890        assert!(normalize_metadata_key("1source").is_err());
891        assert!(normalize_metadata_key("source id").is_err());
892        assert!(normalize_metadata_key("aven.internal").is_err());
893    }
894
895    #[test]
896    fn metadata_result_limits_include_existing_value_bytes() {
897        let mut existing = (0..7)
898            .map(|index| (format!("key_{index}"), "x".repeat(MAX_METADATA_VALUE_BYTES)))
899            .collect::<Vec<_>>();
900        existing.push((
901            "key_7".to_string(),
902            "x".repeat(MAX_METADATA_VALUE_BYTES - 1),
903        ));
904        let set = [TaskMetadataInput {
905            key: "key_8".to_string(),
906            value: "é".to_string(),
907        }];
908
909        let error = validate_metadata_result_limits(existing, &set, &[]).unwrap_err();
910
911        assert_eq!(
912            error.to_string(),
913            format!("error metadata-values-too-large limit={MAX_METADATA_TOTAL_BYTES}")
914        );
915    }
916
917    #[test]
918    fn metadata_result_limits_apply_removals_before_sets() {
919        let existing = (0..8)
920            .map(|index| (format!("key_{index}"), "x".repeat(MAX_METADATA_VALUE_BYTES)))
921            .collect::<Vec<_>>();
922        let set = [TaskMetadataInput {
923            key: "replacement".to_string(),
924            value: "y".repeat(MAX_METADATA_VALUE_BYTES),
925        }];
926
927        validate_metadata_result_limits(existing, &set, &[" KEY_0 ".to_string()]).unwrap();
928    }
929
930    #[test]
931    fn metadata_result_limits_distinguish_replacement_from_insertion_at_count_limit() {
932        let existing = (0..MAX_METADATA_VALUES)
933            .map(|index| {
934                (
935                    format!("key_{index}"),
936                    "x".repeat(MAX_METADATA_TOTAL_BYTES / MAX_METADATA_VALUES),
937                )
938            })
939            .collect::<Vec<_>>();
940        let replacement = [TaskMetadataInput {
941            key: "KEY_0".to_string(),
942            value: "replacement".to_string(),
943        }];
944        validate_metadata_result_limits(existing.clone(), &replacement, &[]).unwrap();
945
946        let insertion = [TaskMetadataInput {
947            key: "extra".to_string(),
948            value: "x".to_string(),
949        }];
950        let error = validate_metadata_result_limits(existing, &insertion, &[]).unwrap_err();
951
952        assert_eq!(
953            error.to_string(),
954            format!("error too-many-metadata-values limit={MAX_METADATA_VALUES}")
955        );
956    }
957
958    #[tokio::test]
959    async fn repeated_key_resolution_reuses_stable_field_identity() {
960        let (_temp, mut conn) = test_conn().await;
961        let workspace = ensure_default_workspace(&mut conn).await.unwrap();
962        let mut tx = begin_immediate(&mut conn).await.unwrap();
963        let first = resolve_or_create_metadata_field(&mut tx, &workspace, "max_turns")
964            .await
965            .unwrap();
966        let second = resolve_or_create_metadata_field(&mut tx, &workspace, "MAX_TURNS")
967            .await
968            .unwrap();
969        tx.commit().await.unwrap();
970
971        assert_eq!(first.id, second.id);
972        let changes: i64 = sqlx::query_scalar(
973            "SELECT count(*) FROM changes WHERE op_type = 'create_metadata_field'",
974        )
975        .fetch_one(&mut *conn)
976        .await
977        .unwrap();
978        assert_eq!(changes, 1);
979    }
980}
aven_core/metadata.rs

aven_core/
metadata.rs
