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 deleted = sqlx::query(
426        "DELETE 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    .execute(&mut *conn)
433    .await?
434    .rows_affected();
435    if deleted == 0 {
436        return Ok(false);
437    }
438    let timestamp = now();
439    touch_task(conn, &workspace.id, task_id, &timestamp).await?;
440    let change_id = insert_change(
441        conn,
442        "task",
443        task_id,
444        Some(&identity),
445        "remove_task_metadata",
446        json!({
447            "workspace_id": &workspace.id,
448            "workspace_key": &workspace.key,
449            "field_id": &field.id,
450            "key": &field.key,
451        }),
452        base.as_deref(),
453    )
454    .await?;
455    set_field_version(conn, task_id, &identity, &change_id).await?;
456    Ok(true)
457}
458
459pub(crate) async fn set_recurrence_metadata(
460    conn: &mut SqliteConnection,
461    workspace: &Workspace,
462    series_id: &RecurrenceSeriesId,
463    input: &TaskMetadataInput,
464) -> Result<bool> {
465    let field = resolve_or_create_metadata_field(conn, workspace, &input.key).await?;
466    let identity = format!("metadata:{}", field.id);
467    let current = sqlx::query_scalar::<_, String>(
468        "SELECT value FROM recurrence_series_metadata
469         WHERE workspace_id = ? AND series_id = ? AND field_id = ?",
470    )
471    .bind(&workspace.id)
472    .bind(series_id)
473    .bind(&field.id)
474    .fetch_optional(&mut *conn)
475    .await?;
476    if current.as_deref() == Some(input.value.as_str()) {
477        return Ok(false);
478    }
479    if entity_conflict_exists(
480        conn,
481        &workspace.id,
482        MutableEntityType::RecurrenceSeries,
483        series_id.as_str(),
484        &identity,
485    )
486    .await?
487    {
488        bail!("error conflicted-recurrence-field field={identity}");
489    }
490    let base = entity_field_version(
491        conn,
492        &workspace.id,
493        MutableEntityType::RecurrenceSeries,
494        series_id.as_str(),
495        &identity,
496    )
497    .await?;
498    let timestamp = now();
499    sqlx::query(
500        "INSERT INTO recurrence_series_metadata(
501             workspace_id, series_id, field_id, value, created_at, updated_at
502         ) VALUES (?, ?, ?, ?, ?, ?)
503         ON CONFLICT(workspace_id, series_id, field_id)
504         DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
505    )
506    .bind(&workspace.id)
507    .bind(series_id)
508    .bind(&field.id)
509    .bind(&input.value)
510    .bind(&timestamp)
511    .bind(&timestamp)
512    .execute(&mut *conn)
513    .await?;
514    sqlx::query("UPDATE recurrence_series SET updated_at = ? WHERE workspace_id = ? AND id = ?")
515        .bind(&timestamp)
516        .bind(&workspace.id)
517        .bind(series_id)
518        .execute(&mut *conn)
519        .await?;
520    let change_id = insert_change(
521        conn,
522        "recurrence_series",
523        series_id.as_str(),
524        Some(&identity),
525        "set_recurrence_metadata",
526        json!({
527            "workspace_id": &workspace.id,
528            "workspace_key": &workspace.key,
529            "field_id": &field.id,
530            "key": &field.key,
531            "value": &input.value,
532        }),
533        base.as_deref(),
534    )
535    .await?;
536    set_entity_field_version(
537        conn,
538        &workspace.id,
539        MutableEntityType::RecurrenceSeries,
540        series_id.as_str(),
541        &identity,
542        &change_id,
543    )
544    .await?;
545    Ok(true)
546}
547
548pub(crate) async fn remove_recurrence_metadata(
549    conn: &mut SqliteConnection,
550    workspace: &Workspace,
551    series_id: &RecurrenceSeriesId,
552    key: &str,
553) -> Result<bool> {
554    let Some(field) =
555        metadata_field_by_key(conn, &workspace.id, &normalize_metadata_key(key)?).await?
556    else {
557        return Ok(false);
558    };
559    let identity = format!("metadata:{}", field.id);
560    if entity_conflict_exists(
561        conn,
562        &workspace.id,
563        MutableEntityType::RecurrenceSeries,
564        series_id.as_str(),
565        &identity,
566    )
567    .await?
568    {
569        bail!("error conflicted-recurrence-field field={identity}");
570    }
571    let removed = sqlx::query(
572        "DELETE FROM recurrence_series_metadata
573         WHERE workspace_id = ? AND series_id = ? AND field_id = ?",
574    )
575    .bind(&workspace.id)
576    .bind(series_id)
577    .bind(&field.id)
578    .execute(&mut *conn)
579    .await?
580    .rows_affected();
581    if removed == 0 {
582        return Ok(false);
583    }
584    let base = entity_field_version(
585        conn,
586        &workspace.id,
587        MutableEntityType::RecurrenceSeries,
588        series_id.as_str(),
589        &identity,
590    )
591    .await?;
592    let timestamp = now();
593    sqlx::query("UPDATE recurrence_series SET updated_at = ? WHERE workspace_id = ? AND id = ?")
594        .bind(&timestamp)
595        .bind(&workspace.id)
596        .bind(series_id)
597        .execute(&mut *conn)
598        .await?;
599    let change_id = insert_change(
600        conn,
601        "recurrence_series",
602        series_id.as_str(),
603        Some(&identity),
604        "remove_recurrence_metadata",
605        json!({
606            "workspace_id": &workspace.id,
607            "workspace_key": &workspace.key,
608            "field_id": &field.id,
609            "key": &field.key,
610        }),
611        base.as_deref(),
612    )
613    .await?;
614    set_entity_field_version(
615        conn,
616        &workspace.id,
617        MutableEntityType::RecurrenceSeries,
618        series_id.as_str(),
619        &identity,
620        &change_id,
621    )
622    .await?;
623    Ok(true)
624}
625
626pub(crate) fn validate_metadata_update(set: &[TaskMetadataInput], remove: &[String]) -> Result<()> {
627    if set.len() > MAX_METADATA_VALUES {
628        bail!("error too-many-metadata-values limit={MAX_METADATA_VALUES}");
629    }
630    let mut keys = BTreeSet::new();
631    let mut total_bytes = 0usize;
632    for input in set {
633        let key = normalize_metadata_key(&input.key)?;
634        if !keys.insert(key) {
635            bail!("error duplicate-metadata-key");
636        }
637        if input.value.len() > MAX_METADATA_VALUE_BYTES {
638            bail!("error metadata-value-too-large limit={MAX_METADATA_VALUE_BYTES}");
639        }
640        total_bytes = total_bytes.saturating_add(input.value.len());
641    }
642    if total_bytes > MAX_METADATA_TOTAL_BYTES {
643        bail!("error metadata-values-too-large limit={MAX_METADATA_TOTAL_BYTES}");
644    }
645    for input in remove {
646        let key = normalize_metadata_key(input)?;
647        if !keys.insert(key) {
648            bail!("error overlapping-metadata-key");
649        }
650    }
651    Ok(())
652}
653
654fn validate_metadata_result_limits(
655    existing: impl IntoIterator<Item = (String, String)>,
656    set: &[TaskMetadataInput],
657    remove: &[String],
658) -> Result<()> {
659    let mut values = existing.into_iter().collect::<HashMap<_, _>>();
660    for key in remove {
661        values.remove(&normalize_metadata_key(key)?);
662    }
663    for input in set {
664        values.insert(normalize_metadata_key(&input.key)?, input.value.clone());
665    }
666    if values.len() > MAX_METADATA_VALUES {
667        bail!("error too-many-metadata-values limit={MAX_METADATA_VALUES}");
668    }
669    let total_bytes = values.values().map(String::len).sum::<usize>();
670    if total_bytes > MAX_METADATA_TOTAL_BYTES {
671        bail!("error metadata-values-too-large limit={MAX_METADATA_TOTAL_BYTES}");
672    }
673    Ok(())
674}
675
676pub(crate) async fn validate_recurrence_metadata_result(
677    conn: &mut SqliteConnection,
678    workspace_id: &WorkspaceId,
679    series_id: &RecurrenceSeriesId,
680    set: &[TaskMetadataInput],
681    remove: &[String],
682) -> Result<()> {
683    validate_metadata_update(set, remove)?;
684    let rows = sqlx::query(
685        "SELECT f.key, m.value FROM recurrence_series_metadata m
686         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
687         WHERE m.workspace_id = ? AND m.series_id = ?",
688    )
689    .bind(workspace_id)
690    .bind(series_id)
691    .fetch_all(&mut *conn)
692    .await?;
693    validate_metadata_result_limits(
694        rows.into_iter()
695            .map(|row| (row.get::<String, _>("key"), row.get::<String, _>("value"))),
696        set,
697        remove,
698    )
699}
700
701pub(crate) async fn validate_task_metadata_result(
702    conn: &mut SqliteConnection,
703    workspace_id: &WorkspaceId,
704    task_id: &TaskId,
705    set: &[TaskMetadataInput],
706    remove: &[String],
707) -> Result<()> {
708    validate_metadata_update(set, remove)?;
709    let rows = sqlx::query(
710        "SELECT f.key, m.value FROM task_metadata m
711         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
712         WHERE m.workspace_id = ? AND m.task_id = ?",
713    )
714    .bind(workspace_id)
715    .bind(task_id)
716    .fetch_all(&mut *conn)
717    .await?;
718    validate_metadata_result_limits(
719        rows.into_iter()
720            .map(|row| (row.get::<String, _>("key"), row.get::<String, _>("value"))),
721        set,
722        remove,
723    )
724}
725
726pub(crate) async fn resolve_metadata_inputs(
727    conn: &mut SqliteConnection,
728    workspace: &Workspace,
729    inputs: &[TaskMetadataInput],
730) -> Result<Vec<ResolvedMetadataValue>> {
731    validate_metadata_update(inputs, &[])?;
732    let mut values = Vec::with_capacity(inputs.len());
733    for input in inputs {
734        let field = resolve_or_create_metadata_field(conn, workspace, &input.key).await?;
735        values.push(ResolvedMetadataValue {
736            field_id: field.id,
737            key: field.key,
738            value: input.value.clone(),
739        });
740    }
741    Ok(values)
742}
743
744async fn touch_task(
745    conn: &mut SqliteConnection,
746    workspace_id: &WorkspaceId,
747    task_id: &TaskId,
748    timestamp: &str,
749) -> Result<()> {
750    let affected = sqlx::query("UPDATE tasks SET updated_at = ? WHERE workspace_id = ? AND id = ?")
751        .bind(timestamp)
752        .bind(workspace_id)
753        .bind(task_id)
754        .execute(&mut *conn)
755        .await?
756        .rows_affected();
757    if affected != 1 {
758        bail!("error task-not-found task_id={task_id}");
759    }
760    Ok(())
761}
762
763pub(crate) async fn task_metadata_in_workspace(
764    conn: &mut SqliteConnection,
765    workspace_id: &WorkspaceId,
766    task_id: &TaskId,
767) -> Result<Vec<TaskMetadataValue>> {
768    let rows = sqlx::query(
769        "SELECT m.field_id, f.key, m.value
770         FROM task_metadata m
771         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
772         WHERE m.workspace_id = ? AND m.task_id = ?
773         ORDER BY f.key",
774    )
775    .bind(workspace_id)
776    .bind(task_id)
777    .fetch_all(&mut *conn)
778    .await?;
779    Ok(rows
780        .into_iter()
781        .map(|row| TaskMetadataValue {
782            field_id: row.get("field_id"),
783            key: row.get("key"),
784            value: row.get("value"),
785        })
786        .collect())
787}
788
789pub(crate) async fn metadata_by_task_ids(
790    conn: &mut SqliteConnection,
791    workspace_id: &WorkspaceId,
792    task_ids: &[TaskId],
793) -> Result<HashMap<TaskId, Vec<TaskMetadataValue>>> {
794    let mut values = HashMap::new();
795    for chunk in task_ids.chunks(900) {
796        let mut query = QueryBuilder::<Sqlite>::new(
797            "SELECT m.task_id, m.field_id, f.key, m.value
798             FROM task_metadata m
799             JOIN metadata_fields f
800               ON f.workspace_id = m.workspace_id AND f.id = m.field_id
801             WHERE m.workspace_id = ",
802        );
803        query.push_bind(workspace_id);
804        query.push(" AND m.task_id IN (");
805        {
806            let mut separated = query.separated(", ");
807            for task_id in chunk {
808                separated.push_bind(task_id);
809            }
810        }
811        query.push(") ORDER BY m.task_id, f.key");
812        for row in query.build().fetch_all(&mut *conn).await? {
813            let task_id: TaskId = row.get("task_id");
814            values
815                .entry(task_id)
816                .or_insert_with(Vec::new)
817                .push(TaskMetadataValue {
818                    field_id: row.get("field_id"),
819                    key: row.get("key"),
820                    value: row.get("value"),
821                });
822        }
823    }
824    Ok(values)
825}
826
827async fn list_metadata_fields_in_workspace(
828    conn: &mut SqliteConnection,
829    workspace_id: &WorkspaceId,
830) -> Result<Vec<MetadataFieldUsage>> {
831    let rows = sqlx::query(
832        "SELECT f.id, f.workspace_id, f.key, f.created_at, f.updated_at,
833                (SELECT count(*) FROM task_metadata m
834                 WHERE m.workspace_id = f.workspace_id AND m.field_id = f.id) AS task_count,
835                (SELECT count(*) FROM recurrence_series_metadata m
836                 WHERE m.workspace_id = f.workspace_id AND m.field_id = f.id) AS series_count
837         FROM metadata_fields f
838         WHERE f.workspace_id = ?
839         ORDER BY f.key",
840    )
841    .bind(workspace_id)
842    .fetch_all(&mut *conn)
843    .await?;
844    Ok(rows
845        .into_iter()
846        .map(|row| MetadataFieldUsage {
847            field: MetadataField {
848                id: row.get("id"),
849                workspace_id: row.get("workspace_id"),
850                key: row.get("key"),
851                created_at: row.get("created_at"),
852                updated_at: row.get("updated_at"),
853            },
854            task_count: usize::try_from(row.get::<i64, _>("task_count")).unwrap_or(usize::MAX),
855            series_count: usize::try_from(row.get::<i64, _>("series_count")).unwrap_or(usize::MAX),
856        })
857        .collect())
858}
859
860fn metadata_field_from_row(row: sqlx::sqlite::SqliteRow) -> MetadataField {
861    MetadataField {
862        id: row.get("id"),
863        workspace_id: row.get("workspace_id"),
864        key: row.get("key"),
865        created_at: row.get("created_at"),
866        updated_at: row.get("updated_at"),
867    }
868}
869
870#[cfg(test)]
871mod tests {
872    use super::*;
873    use crate::db::begin_immediate;
874    use crate::test_support::{ensure_default_workspace, test_conn};
875
876    #[test]
877    fn metadata_keys_have_stable_normalization() {
878        assert_eq!(normalize_metadata_key(" Max_Turns ").unwrap(), "max_turns");
879        assert_eq!(normalize_metadata_key("source.id").unwrap(), "source.id");
880        assert!(normalize_metadata_key("1source").is_err());
881        assert!(normalize_metadata_key("source id").is_err());
882        assert!(normalize_metadata_key("aven.internal").is_err());
883    }
884
885    #[test]
886    fn metadata_result_limits_include_existing_value_bytes() {
887        let mut existing = (0..7)
888            .map(|index| (format!("key_{index}"), "x".repeat(MAX_METADATA_VALUE_BYTES)))
889            .collect::<Vec<_>>();
890        existing.push((
891            "key_7".to_string(),
892            "x".repeat(MAX_METADATA_VALUE_BYTES - 1),
893        ));
894        let set = [TaskMetadataInput {
895            key: "key_8".to_string(),
896            value: "é".to_string(),
897        }];
898
899        let error = validate_metadata_result_limits(existing, &set, &[]).unwrap_err();
900
901        assert_eq!(
902            error.to_string(),
903            format!("error metadata-values-too-large limit={MAX_METADATA_TOTAL_BYTES}")
904        );
905    }
906
907    #[test]
908    fn metadata_result_limits_apply_removals_before_sets() {
909        let existing = (0..8)
910            .map(|index| (format!("key_{index}"), "x".repeat(MAX_METADATA_VALUE_BYTES)))
911            .collect::<Vec<_>>();
912        let set = [TaskMetadataInput {
913            key: "replacement".to_string(),
914            value: "y".repeat(MAX_METADATA_VALUE_BYTES),
915        }];
916
917        validate_metadata_result_limits(existing, &set, &[" KEY_0 ".to_string()]).unwrap();
918    }
919
920    #[test]
921    fn metadata_result_limits_distinguish_replacement_from_insertion_at_count_limit() {
922        let existing = (0..MAX_METADATA_VALUES)
923            .map(|index| {
924                (
925                    format!("key_{index}"),
926                    "x".repeat(MAX_METADATA_TOTAL_BYTES / MAX_METADATA_VALUES),
927                )
928            })
929            .collect::<Vec<_>>();
930        let replacement = [TaskMetadataInput {
931            key: "KEY_0".to_string(),
932            value: "replacement".to_string(),
933        }];
934        validate_metadata_result_limits(existing.clone(), &replacement, &[]).unwrap();
935
936        let insertion = [TaskMetadataInput {
937            key: "extra".to_string(),
938            value: "x".to_string(),
939        }];
940        let error = validate_metadata_result_limits(existing, &insertion, &[]).unwrap_err();
941
942        assert_eq!(
943            error.to_string(),
944            format!("error too-many-metadata-values limit={MAX_METADATA_VALUES}")
945        );
946    }
947
948    #[tokio::test]
949    async fn repeated_key_resolution_reuses_stable_field_identity() {
950        let (_temp, mut conn) = test_conn().await;
951        let workspace = ensure_default_workspace(&mut conn).await.unwrap();
952        let mut tx = begin_immediate(&mut conn).await.unwrap();
953        let first = resolve_or_create_metadata_field(&mut tx, &workspace, "max_turns")
954            .await
955            .unwrap();
956        let second = resolve_or_create_metadata_field(&mut tx, &workspace, "MAX_TURNS")
957            .await
958            .unwrap();
959        tx.commit().await.unwrap();
960
961        assert_eq!(first.id, second.id);
962        let changes: i64 = sqlx::query_scalar(
963            "SELECT count(*) FROM changes WHERE op_type = 'create_metadata_field'",
964        )
965        .fetch_one(&mut *conn)
966        .await
967        .unwrap();
968        assert_eq!(changes, 1);
969    }
970}
aven_core/metadata.rs

aven_core/
metadata.rs
