ai_crew_sync/store/
notes.rs

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

ai_crew_sync/store/
notes.rs
