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

aven_core/
metadata.rs
