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
654pub(crate) async fn validate_recurrence_metadata_result(
655    conn: &mut SqliteConnection,
656    workspace_id: &WorkspaceId,
657    series_id: &RecurrenceSeriesId,
658    set: &[TaskMetadataInput],
659    remove: &[String],
660) -> Result<()> {
661    validate_metadata_update(set, remove)?;
662    let rows = sqlx::query(
663        "SELECT f.key, m.value FROM recurrence_series_metadata m
664         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
665         WHERE m.workspace_id = ? AND m.series_id = ?",
666    )
667    .bind(workspace_id)
668    .bind(series_id)
669    .fetch_all(&mut *conn)
670    .await?;
671    let mut values = rows
672        .into_iter()
673        .map(|row| (row.get::<String, _>("key"), row.get::<String, _>("value")))
674        .collect::<HashMap<_, _>>();
675    for key in remove {
676        values.remove(&normalize_metadata_key(key)?);
677    }
678    for input in set {
679        values.insert(normalize_metadata_key(&input.key)?, input.value.clone());
680    }
681    if values.len() > MAX_METADATA_VALUES {
682        bail!("error too-many-metadata-values limit={MAX_METADATA_VALUES}");
683    }
684    let total_bytes = values.values().map(String::len).sum::<usize>();
685    if total_bytes > MAX_METADATA_TOTAL_BYTES {
686        bail!("error metadata-values-too-large limit={MAX_METADATA_TOTAL_BYTES}");
687    }
688    Ok(())
689}
690
691pub(crate) async fn validate_task_metadata_result(
692    conn: &mut SqliteConnection,
693    workspace_id: &WorkspaceId,
694    task_id: &TaskId,
695    set: &[TaskMetadataInput],
696    remove: &[String],
697) -> Result<()> {
698    validate_metadata_update(set, remove)?;
699    let rows = sqlx::query(
700        "SELECT f.key, m.value FROM task_metadata m
701         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
702         WHERE m.workspace_id = ? AND m.task_id = ?",
703    )
704    .bind(workspace_id)
705    .bind(task_id)
706    .fetch_all(&mut *conn)
707    .await?;
708    let mut values = rows
709        .into_iter()
710        .map(|row| (row.get::<String, _>("key"), row.get::<String, _>("value")))
711        .collect::<HashMap<_, _>>();
712    for key in remove {
713        values.remove(&normalize_metadata_key(key)?);
714    }
715    for input in set {
716        values.insert(normalize_metadata_key(&input.key)?, input.value.clone());
717    }
718    if values.len() > MAX_METADATA_VALUES {
719        bail!("error too-many-metadata-values limit={MAX_METADATA_VALUES}");
720    }
721    let total_bytes = values.values().map(String::len).sum::<usize>();
722    if total_bytes > MAX_METADATA_TOTAL_BYTES {
723        bail!("error metadata-values-too-large limit={MAX_METADATA_TOTAL_BYTES}");
724    }
725    Ok(())
726}
727
728pub(crate) async fn resolve_metadata_inputs(
729    conn: &mut SqliteConnection,
730    workspace: &Workspace,
731    inputs: &[TaskMetadataInput],
732) -> Result<Vec<ResolvedMetadataValue>> {
733    validate_metadata_update(inputs, &[])?;
734    let mut values = Vec::with_capacity(inputs.len());
735    for input in inputs {
736        let field = resolve_or_create_metadata_field(conn, workspace, &input.key).await?;
737        values.push(ResolvedMetadataValue {
738            field_id: field.id,
739            key: field.key,
740            value: input.value.clone(),
741        });
742    }
743    Ok(values)
744}
745
746async fn touch_task(
747    conn: &mut SqliteConnection,
748    workspace_id: &WorkspaceId,
749    task_id: &TaskId,
750    timestamp: &str,
751) -> Result<()> {
752    let affected = sqlx::query("UPDATE tasks SET updated_at = ? WHERE workspace_id = ? AND id = ?")
753        .bind(timestamp)
754        .bind(workspace_id)
755        .bind(task_id)
756        .execute(&mut *conn)
757        .await?
758        .rows_affected();
759    if affected != 1 {
760        bail!("error task-not-found task_id={task_id}");
761    }
762    Ok(())
763}
764
765pub(crate) async fn task_metadata_in_workspace(
766    conn: &mut SqliteConnection,
767    workspace_id: &WorkspaceId,
768    task_id: &TaskId,
769) -> Result<Vec<TaskMetadataValue>> {
770    let rows = sqlx::query(
771        "SELECT m.field_id, f.key, m.value
772         FROM task_metadata m
773         JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id
774         WHERE m.workspace_id = ? AND m.task_id = ?
775         ORDER BY f.key",
776    )
777    .bind(workspace_id)
778    .bind(task_id)
779    .fetch_all(&mut *conn)
780    .await?;
781    Ok(rows
782        .into_iter()
783        .map(|row| TaskMetadataValue {
784            field_id: row.get("field_id"),
785            key: row.get("key"),
786            value: row.get("value"),
787        })
788        .collect())
789}
790
791pub(crate) async fn metadata_by_task_ids(
792    conn: &mut SqliteConnection,
793    workspace_id: &WorkspaceId,
794    task_ids: &[TaskId],
795) -> Result<HashMap<TaskId, Vec<TaskMetadataValue>>> {
796    let mut values = HashMap::new();
797    for chunk in task_ids.chunks(900) {
798        let mut query = QueryBuilder::<Sqlite>::new(
799            "SELECT m.task_id, m.field_id, f.key, m.value
800             FROM task_metadata m
801             JOIN metadata_fields f
802               ON f.workspace_id = m.workspace_id AND f.id = m.field_id
803             WHERE m.workspace_id = ",
804        );
805        query.push_bind(workspace_id);
806        query.push(" AND m.task_id IN (");
807        {
808            let mut separated = query.separated(", ");
809            for task_id in chunk {
810                separated.push_bind(task_id);
811            }
812        }
813        query.push(") ORDER BY m.task_id, f.key");
814        for row in query.build().fetch_all(&mut *conn).await? {
815            let task_id: TaskId = row.get("task_id");
816            values
817                .entry(task_id)
818                .or_insert_with(Vec::new)
819                .push(TaskMetadataValue {
820                    field_id: row.get("field_id"),
821                    key: row.get("key"),
822                    value: row.get("value"),
823                });
824        }
825    }
826    Ok(values)
827}
828
829async fn list_metadata_fields_in_workspace(
830    conn: &mut SqliteConnection,
831    workspace_id: &WorkspaceId,
832) -> Result<Vec<MetadataFieldUsage>> {
833    let rows = sqlx::query(
834        "SELECT f.id, f.workspace_id, f.key, f.created_at, f.updated_at,
835                (SELECT count(*) FROM task_metadata m
836                 WHERE m.workspace_id = f.workspace_id AND m.field_id = f.id) AS task_count,
837                (SELECT count(*) FROM recurrence_series_metadata m
838                 WHERE m.workspace_id = f.workspace_id AND m.field_id = f.id) AS series_count
839         FROM metadata_fields f
840         WHERE f.workspace_id = ?
841         ORDER BY f.key",
842    )
843    .bind(workspace_id)
844    .fetch_all(&mut *conn)
845    .await?;
846    Ok(rows
847        .into_iter()
848        .map(|row| MetadataFieldUsage {
849            field: MetadataField {
850                id: row.get("id"),
851                workspace_id: row.get("workspace_id"),
852                key: row.get("key"),
853                created_at: row.get("created_at"),
854                updated_at: row.get("updated_at"),
855            },
856            task_count: usize::try_from(row.get::<i64, _>("task_count")).unwrap_or(usize::MAX),
857            series_count: usize::try_from(row.get::<i64, _>("series_count")).unwrap_or(usize::MAX),
858        })
859        .collect())
860}
861
862fn metadata_field_from_row(row: sqlx::sqlite::SqliteRow) -> MetadataField {
863    MetadataField {
864        id: row.get("id"),
865        workspace_id: row.get("workspace_id"),
866        key: row.get("key"),
867        created_at: row.get("created_at"),
868        updated_at: row.get("updated_at"),
869    }
870}
871
872#[cfg(test)]
873mod tests {
874    use super::*;
875    use crate::db::begin_immediate;
876    use crate::test_support::{ensure_default_workspace, test_conn};
877
878    #[test]
879    fn metadata_keys_have_stable_normalization() {
880        assert_eq!(normalize_metadata_key(" Max_Turns ").unwrap(), "max_turns");
881        assert_eq!(normalize_metadata_key("source.id").unwrap(), "source.id");
882        assert!(normalize_metadata_key("1source").is_err());
883        assert!(normalize_metadata_key("source id").is_err());
884        assert!(normalize_metadata_key("aven.internal").is_err());
885    }
886
887    #[tokio::test]
888    async fn repeated_key_resolution_reuses_stable_field_identity() {
889        let (_temp, mut conn) = test_conn().await;
890        let workspace = ensure_default_workspace(&mut conn).await.unwrap();
891        let mut tx = begin_immediate(&mut conn).await.unwrap();
892        let first = resolve_or_create_metadata_field(&mut tx, &workspace, "max_turns")
893            .await
894            .unwrap();
895        let second = resolve_or_create_metadata_field(&mut tx, &workspace, "MAX_TURNS")
896            .await
897            .unwrap();
898        tx.commit().await.unwrap();
899
900        assert_eq!(first.id, second.id);
901        let changes: i64 = sqlx::query_scalar(
902            "SELECT count(*) FROM changes WHERE op_type = 'create_metadata_field'",
903        )
904        .fetch_one(&mut *conn)
905        .await
906        .unwrap();
907        assert_eq!(changes, 1);
908    }
909}
aven_core/metadata.rs

aven_core/
metadata.rs
