ai_crew_sync/store/
notes.rs

1use sqlx::PgPool;
2use uuid::Uuid;
3
4use crate::{
5    auth::AuthCtx,
6    error::{BusError, BusResult},
7    model::{Ack, NoteInfo, NoteList, NoteRef, ts},
8};
9
10/// A note is the team's durable memory: runbooks and decision records
11/// belong here whole, not truncated.
12const MAX_VALUE_BYTES: usize = 1024 * 1024;
13/// Tags are filters, not content: a bounded handful of short labels.
14const MAX_TAGS: usize = 16;
15const MAX_TAG_BYTES: usize = 64;
16const MAX_LIMIT: i64 = 200;
17
18#[derive(sqlx::FromRow)]
19struct NoteRow {
20    scope: String,
21    key: String,
22    value: String,
23    tags: Vec<String>,
24    updated_by: Option<String>,
25    updated_at: chrono::DateTime<chrono::Utc>,
26}
27
28impl From<NoteRow> for NoteInfo {
29    fn from(r: NoteRow) -> Self {
30        NoteInfo {
31            scope: r.scope,
32            key: r.key,
33            value: r.value,
34            tags: r.tags,
35            updated_by: r.updated_by,
36            updated_at: ts(r.updated_at),
37        }
38    }
39}
40
41const NOTE_SELECT: &str = r#"
42    SELECT n.scope, n.key, n.value, n.tags, a.name AS updated_by, n.updated_at
43    FROM notes n
44    LEFT JOIN agents a ON a.id = n.updated_by
45"#;
46
47fn normalize_scope(scope: Option<String>) -> String {
48    scope
49        .map(|s| s.trim().to_lowercase())
50        .filter(|s| !s.is_empty())
51        .unwrap_or_else(|| "global".into())
52}
53
54pub struct SetInput {
55    pub scope: Option<String>,
56    pub key: String,
57    pub value: String,
58    pub tags: Option<Vec<String>>,
59}
60
61pub async fn set_note(pool: &PgPool, auth: &AuthCtx, input: SetInput) -> BusResult<NoteInfo> {
62    let scope = normalize_scope(input.scope);
63    let key = input.key.trim().to_owned();
64    if key.is_empty() {
65        return Err(BusError::invalid("note key cannot be empty"));
66    }
67    if input.value.len() > MAX_VALUE_BYTES {
68        return Err(BusError::invalid(format!(
69            "note value is {} bytes; the limit is {MAX_VALUE_BYTES}",
70            input.value.len()
71        )));
72    }
73    let tags: Vec<String> = input
74        .tags
75        .unwrap_or_default()
76        .into_iter()
77        .map(|t| t.trim().to_lowercase())
78        .filter(|t| !t.is_empty())
79        .collect();
80    if tags.len() > MAX_TAGS {
81        return Err(BusError::invalid(format!(
82            "a note carries at most {MAX_TAGS} tags; got {}",
83            tags.len()
84        )));
85    }
86    for tag in &tags {
87        super::check_text("note tag", tag, MAX_TAG_BYTES)?;
88    }
89
90    let (id,): (Uuid,) = sqlx::query_as(
91        r#"
92        INSERT INTO notes (team_id, scope, key, value, tags, updated_by)
93        VALUES ($1, $2, $3, $4, $5, $6)
94        ON CONFLICT (team_id, scope, key) DO UPDATE SET
95            value = EXCLUDED.value,
96            tags = EXCLUDED.tags,
97            updated_by = EXCLUDED.updated_by,
98            updated_at = now()
99        RETURNING id
100        "#,
101    )
102    .bind(auth.team_id)
103    .bind(&scope)
104    .bind(&key)
105    .bind(&input.value)
106    .bind(&tags)
107    .bind(auth.agent_id)
108    .fetch_one(pool)
109    .await?;
110
111    // Keep an append-only trail so a bad overwrite is recoverable.
112    sqlx::query("INSERT INTO note_revisions (note_id, value, updated_by) VALUES ($1, $2, $3)")
113        .bind(id)
114        .bind(&input.value)
115        .bind(auth.agent_id)
116        .execute(pool)
117        .await?;
118
119    get_note(pool, auth, Some(scope), &key)
120        .await?
121        .note
122        .ok_or_else(|| BusError::not_found("note just written"))
123}
124
125pub async fn get_note(
126    pool: &PgPool,
127    auth: &AuthCtx,
128    scope: Option<String>,
129    key: &str,
130) -> BusResult<NoteRef> {
131    let scope = normalize_scope(scope);
132    let key = key.trim().to_owned();
133
134    let row: Option<NoteRow> = sqlx::query_as(&format!(
135        "{NOTE_SELECT} WHERE n.team_id = $1 AND n.scope = $2 AND n.key = $3"
136    ))
137    .bind(auth.team_id)
138    .bind(&scope)
139    .bind(&key)
140    .fetch_optional(pool)
141    .await?;
142
143    Ok(NoteRef {
144        scope,
145        key,
146        found: row.is_some(),
147        note: row.map(Into::into),
148    })
149}
150
151pub async fn list_notes(
152    pool: &PgPool,
153    auth: &AuthCtx,
154    scope: Option<String>,
155    tag: Option<String>,
156    limit: i64,
157) -> BusResult<NoteList> {
158    let limit = limit.clamp(1, MAX_LIMIT);
159    // `scope = None` means "every scope", so it is not normalised to 'global'.
160    let scope = scope
161        .map(|s| s.trim().to_lowercase())
162        .filter(|s| !s.is_empty());
163    let tag = tag
164        .map(|t| t.trim().to_lowercase())
165        .filter(|t| !t.is_empty());
166
167    let rows: Vec<NoteRow> = sqlx::query_as(&format!(
168        r#"{NOTE_SELECT}
169           WHERE n.team_id = $1
170             AND ($2::text IS NULL OR n.scope = $2)
171             AND ($3::text IS NULL OR $3 = ANY(n.tags))
172           ORDER BY n.scope, n.key
173           LIMIT $4"#
174    ))
175    .bind(auth.team_id)
176    .bind(scope.as_deref())
177    .bind(tag.as_deref())
178    .bind(limit)
179    .fetch_all(pool)
180    .await?;
181
182    Ok(NoteList {
183        notes: rows.into_iter().map(Into::into).collect(),
184    })
185}
186
187pub async fn search_notes(
188    pool: &PgPool,
189    auth: &AuthCtx,
190    query: &str,
191    scope: Option<String>,
192    limit: i64,
193) -> BusResult<NoteList> {
194    let query = query.trim();
195    if query.is_empty() {
196        return Err(BusError::invalid("search query cannot be empty"));
197    }
198    let limit = limit.clamp(1, MAX_LIMIT);
199    let scope = scope
200        .map(|s| s.trim().to_lowercase())
201        .filter(|s| !s.is_empty());
202
203    let rows: Vec<NoteRow> = sqlx::query_as(&format!(
204        r#"{NOTE_SELECT}
205           WHERE n.team_id = $1
206             AND ($2::text IS NULL OR n.scope = $2)
207             AND to_tsvector('simple', n.key || ' ' || n.value)
208                 @@ plainto_tsquery('simple', $3)
209           ORDER BY n.updated_at DESC
210           LIMIT $4"#
211    ))
212    .bind(auth.team_id)
213    .bind(scope.as_deref())
214    .bind(query)
215    .bind(limit)
216    .fetch_all(pool)
217    .await?;
218
219    Ok(NoteList {
220        notes: rows.into_iter().map(Into::into).collect(),
221    })
222}
223
224pub async fn delete_note(
225    pool: &PgPool,
226    auth: &AuthCtx,
227    scope: Option<String>,
228    key: &str,
229) -> BusResult<Ack> {
230    let scope = normalize_scope(scope);
231    let key = key.trim();
232
233    let res = sqlx::query("DELETE FROM notes WHERE team_id = $1 AND scope = $2 AND key = $3")
234        .bind(auth.team_id)
235        .bind(&scope)
236        .bind(key)
237        .execute(pool)
238        .await?;
239
240    Ok(Ack {
241        ok: res.rows_affected() > 0,
242        detail: if res.rows_affected() > 0 {
243            format!("deleted note {scope}/{key}")
244        } else {
245            format!("no note at {scope}/{key}")
246        },
247    })
248}
ai_crew_sync/store/notes.rs

ai_crew_sync/store/
notes.rs
