Skip to main content

khive_db/stores/
note.rs

1//! SQL-backed `NoteStore` implementation.
2
3use std::sync::Arc;
4
5use async_trait::async_trait;
6use uuid::Uuid;
7
8use khive_storage::error::StorageError;
9use khive_storage::note::{FilterOp, Note, NoteFilter, SortDir};
10use khive_storage::types::{BatchWriteSummary, DeleteMode, Page, PageRequest, SqlValue};
11use khive_storage::NoteStore;
12use khive_storage::StorageCapability;
13
14use crate::error::SqliteError;
15use crate::pool::ConnectionPool;
16
17fn map_err(e: rusqlite::Error, op: &'static str) -> StorageError {
18    StorageError::driver(StorageCapability::Notes, op, e)
19}
20
21fn map_sqlite_err(e: SqliteError, op: &'static str) -> StorageError {
22    StorageError::driver(StorageCapability::Notes, op, e)
23}
24
25/// A NoteStore backed by SQLite. Namespace is the caller's responsibility.
26///
27/// UUID is globally unique — get/delete by ID alone. Query/count use the
28/// namespace parameter as passed. The store is just a pool + is_file_backed.
29pub struct SqlNoteStore {
30    pool: Arc<ConnectionPool>,
31    is_file_backed: bool,
32}
33
34impl SqlNoteStore {
35    /// Create a new store.
36    pub fn new(pool: Arc<ConnectionPool>, is_file_backed: bool) -> Self {
37        Self {
38            pool,
39            is_file_backed,
40        }
41    }
42
43    fn open_standalone_reader(&self) -> Result<rusqlite::Connection, StorageError> {
44        let config = self.pool.config();
45        let path = config.path.as_ref().ok_or_else(|| StorageError::Pool {
46            operation: "note_reader".into(),
47            message: "in-memory databases do not support standalone connections".into(),
48        })?;
49
50        let conn = rusqlite::Connection::open_with_flags(
51            path,
52            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY
53                | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX
54                | rusqlite::OpenFlags::SQLITE_OPEN_URI,
55        )
56        .map_err(|e| map_err(e, "open_note_reader"))?;
57
58        conn.busy_timeout(config.busy_timeout)
59            .map_err(|e| map_err(e, "open_note_reader"))?;
60        conn.pragma_update(None, "foreign_keys", "ON")
61            .map_err(|e| map_err(e, "open_note_reader"))?;
62        conn.pragma_update(None, "synchronous", "NORMAL")
63            .map_err(|e| map_err(e, "open_note_reader"))?;
64
65        Ok(conn)
66    }
67
68    /// Write via pool writer (serializes writes through the mutex).
69    async fn with_writer<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
70    where
71        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
72        R: Send + 'static,
73    {
74        let pool = Arc::clone(&self.pool);
75        tokio::task::spawn_blocking(move || {
76            let guard = pool.try_writer().map_err(|e| map_sqlite_err(e, op))?;
77            f(guard.conn()).map_err(|e| map_err(e, op))
78        })
79        .await
80        .map_err(|e| StorageError::driver(StorageCapability::Notes, op, e))?
81    }
82
83    async fn with_reader<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
84    where
85        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
86        R: Send + 'static,
87    {
88        if self.is_file_backed {
89            let conn = self.open_standalone_reader()?;
90            tokio::task::spawn_blocking(move || f(&conn).map_err(|e| map_err(e, op)))
91                .await
92                .map_err(|e| StorageError::driver(StorageCapability::Notes, op, e))?
93        } else {
94            let pool = Arc::clone(&self.pool);
95            tokio::task::spawn_blocking(move || {
96                let guard = pool.reader().map_err(|e| map_sqlite_err(e, op))?;
97                f(guard.conn()).map_err(|e| map_err(e, op))
98            })
99            .await
100            .map_err(|e| StorageError::driver(StorageCapability::Notes, op, e))?
101        }
102    }
103}
104
105// =============================================================================
106// Helpers
107// =============================================================================
108
109fn read_note(row: &rusqlite::Row<'_>) -> Result<Note, rusqlite::Error> {
110    let id_str: String = row.get(0)?;
111    let namespace: String = row.get(1)?;
112    let kind: String = row.get(2)?;
113    let status: String = row.get(3)?;
114    let name: Option<String> = row.get(4)?;
115    let content: String = row.get(5)?;
116    let salience: Option<f64> = row.get(6)?;
117    let decay_factor: Option<f64> = row.get(7)?;
118    let expires_at: Option<i64> = row.get(8)?;
119    let properties_str: Option<String> = row.get(9)?;
120    let created_at: i64 = row.get(10)?;
121    let updated_at: i64 = row.get(11)?;
122    let deleted_at: Option<i64> = row.get(12)?;
123
124    let id = parse_uuid(&id_str)?;
125
126    let properties = properties_str
127        .map(|s| {
128            serde_json::from_str(&s).map_err(|e| {
129                rusqlite::Error::FromSqlConversionFailure(
130                    9,
131                    rusqlite::types::Type::Text,
132                    Box::new(e),
133                )
134            })
135        })
136        .transpose()?;
137
138    Ok(Note {
139        id,
140        namespace,
141        kind,
142        status,
143        name,
144        content,
145        salience,
146        decay_factor,
147        expires_at,
148        properties,
149        created_at,
150        updated_at,
151        deleted_at,
152    })
153}
154
155fn parse_uuid(s: &str) -> Result<Uuid, rusqlite::Error> {
156    Uuid::parse_str(s).map_err(|e| {
157        rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(e))
158    })
159}
160
161fn build_note_where(
162    namespace: &str,
163    kind: Option<&str>,
164) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
165    let mut conditions: Vec<String> = vec![
166        "namespace = ?1".to_string(),
167        "deleted_at IS NULL".to_string(),
168    ];
169    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(namespace.to_string())];
170
171    if let Some(k) = kind {
172        params.push(Box::new(k.to_string()));
173        conditions.push(format!("kind = ?{}", params.len()));
174    }
175
176    let clause = format!(" WHERE {}", conditions.join(" AND "));
177    (clause, params)
178}
179
180/// Validate that a json_path is safe to interpolate into SQL.
181/// Accepts only `$.field` or `$.field.subfield` paths with alphanumeric/underscore segments.
182fn validate_json_path(path: &str) -> Result<(), StorageError> {
183    let valid = path.starts_with("$.")
184        && path[2..].split('.').all(|part| {
185            !part.is_empty() && part.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
186        });
187    if valid {
188        Ok(())
189    } else {
190        Err(StorageError::InvalidInput {
191            capability: StorageCapability::Notes,
192            operation: "query_notes_filtered".into(),
193            message: format!("invalid JSON path for note filter: {path:?}"),
194        })
195    }
196}
197
198fn json_extract_expr(path: &str) -> String {
199    format!("json_extract(properties, '{path}')")
200}
201
202fn json_type_expr(path: &str) -> String {
203    format!("json_type(properties, '{path}')")
204}
205
206fn sql_value_param(value: &SqlValue) -> Result<Box<dyn rusqlite::types::ToSql>, rusqlite::Error> {
207    Ok(match value {
208        SqlValue::Null => Box::new(Option::<String>::None),
209        SqlValue::Bool(v) => Box::new(*v as i64),
210        SqlValue::Integer(v) => Box::new(*v),
211        SqlValue::Float(v) => Box::new(*v),
212        SqlValue::Text(v) => Box::new(v.clone()),
213        SqlValue::Blob(v) => Box::new(v.clone()),
214        SqlValue::Json(v) => Box::new(
215            serde_json::to_string(v)
216                .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
217        ),
218        SqlValue::Uuid(v) => Box::new(v.to_string()),
219        SqlValue::Timestamp(v) => Box::new(v.timestamp_micros()),
220    })
221}
222
223fn build_note_filter_where(
224    namespace: &str,
225    filter: &NoteFilter,
226) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), rusqlite::Error> {
227    // When filter.namespaces is non-empty use `namespace IN (...)` for
228    // multi-namespace read visibility. Otherwise fall back to equality.
229    let (ns_condition, ns_params): (String, Vec<Box<dyn rusqlite::types::ToSql>>) =
230        if !filter.namespaces.is_empty() {
231            let placeholders: Vec<String> = (1..=filter.namespaces.len())
232                .map(|i| format!("?{i}"))
233                .collect();
234            let params: Vec<Box<dyn rusqlite::types::ToSql>> = filter
235                .namespaces
236                .iter()
237                .map(|ns| -> Box<dyn rusqlite::types::ToSql> { Box::new(ns.clone()) })
238                .collect();
239            (
240                format!("namespace IN ({})", placeholders.join(", ")),
241                params,
242            )
243        } else {
244            (
245                "namespace = ?1".to_string(),
246                vec![Box::new(namespace.to_string())],
247            )
248        };
249
250    let mut conditions = vec![ns_condition, "deleted_at IS NULL".to_string()];
251    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = ns_params;
252
253    if let Some(kind) = &filter.kind {
254        params.push(Box::new(kind.clone()));
255        conditions.push(format!("kind = ?{}", params.len()));
256    }
257
258    for pf in &filter.property_filters {
259        match pf.op {
260            FilterOp::EqOrMissing => {
261                let expr = json_extract_expr(&pf.json_path);
262                params.push(sql_value_param(&pf.value)?);
263                conditions.push(format!(
264                    "({expr} = ?{n} OR {expr} IS NULL)",
265                    n = params.len()
266                ));
267            }
268            FilterOp::JsonTypeEq => {
269                let type_expr = json_type_expr(&pf.json_path);
270                params.push(sql_value_param(&pf.value)?);
271                conditions.push(format!("{type_expr} = ?{}", params.len()));
272            }
273            FilterOp::JsonTypeNeMissing => {
274                let type_expr = json_type_expr(&pf.json_path);
275                params.push(sql_value_param(&pf.value)?);
276                let n = params.len();
277                conditions.push(format!("({type_expr} IS NULL OR {type_expr} != ?{n})"));
278            }
279            _ => {
280                let expr = json_extract_expr(&pf.json_path);
281                let op = match pf.op {
282                    FilterOp::Eq => "=",
283                    FilterOp::Ne => "!=",
284                    FilterOp::Lt => "<",
285                    FilterOp::Lte => "<=",
286                    FilterOp::Gt => ">",
287                    FilterOp::Gte => ">=",
288                    FilterOp::EqOrMissing | FilterOp::JsonTypeEq | FilterOp::JsonTypeNeMissing => {
289                        unreachable!()
290                    }
291                };
292                params.push(sql_value_param(&pf.value)?);
293                conditions.push(format!("{expr} {op} ?{}", params.len()));
294            }
295        }
296    }
297
298    if let Some(min_ts) = filter.min_created_at {
299        params.push(Box::new(min_ts));
300        conditions.push(format!("created_at >= ?{}", params.len()));
301    }
302
303    Ok((format!(" WHERE {}", conditions.join(" AND ")), params))
304}
305
306// =============================================================================
307// NoteStore implementation
308// =============================================================================
309
310#[async_trait]
311impl NoteStore for SqlNoteStore {
312    async fn upsert_note(&self, note: Note) -> Result<(), StorageError> {
313        let namespace = note.namespace.clone();
314        let id_str = note.id.to_string();
315        let kind_str = note.kind.to_string();
316        let status_str = note.status.clone();
317        let properties_str = note
318            .properties
319            .as_ref()
320            .map(|v| serde_json::to_string(v).unwrap_or_default());
321
322        self.with_writer("upsert_note", move |conn| {
323            conn.execute(
324                "INSERT OR REPLACE INTO notes \
325                 (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
326                  properties, created_at, updated_at, deleted_at) \
327                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
328                rusqlite::params![
329                    id_str,
330                    namespace,
331                    kind_str,
332                    status_str,
333                    note.name,
334                    note.content,
335                    note.salience,
336                    note.decay_factor,
337                    note.expires_at,
338                    properties_str,
339                    note.created_at,
340                    note.updated_at,
341                    note.deleted_at,
342                ],
343            )?;
344            Ok(())
345        })
346        .await
347    }
348
349    async fn try_insert_note(&self, note: Note) -> Result<bool, StorageError> {
350        let namespace = note.namespace.clone();
351        let id_str = note.id.to_string();
352        let kind_str = note.kind.to_string();
353        let status_str = note.status.clone();
354        let properties_str = note
355            .properties
356            .as_ref()
357            .map(|v| serde_json::to_string(v).unwrap_or_default());
358
359        // Extract external_id (if any) for dedup verification after a zero-row insert.
360        let ext_id_opt: Option<String> = note
361            .properties
362            .as_ref()
363            .and_then(|v| v.get("external_id"))
364            .and_then(|v| v.as_str())
365            .filter(|s| !s.is_empty())
366            .map(|s| s.to_string());
367
368        self.with_writer("try_insert_note", move |conn| {
369            let rows = conn.execute(
370                "INSERT OR IGNORE INTO notes \
371                 (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
372                  properties, created_at, updated_at, deleted_at) \
373                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
374                rusqlite::params![
375                    id_str,
376                    namespace,
377                    kind_str,
378                    status_str,
379                    note.name,
380                    note.content,
381                    note.salience,
382                    note.decay_factor,
383                    note.expires_at,
384                    properties_str,
385                    note.created_at,
386                    note.updated_at,
387                    note.deleted_at,
388                ],
389            )?;
390
391            if rows > 0 {
392                return Ok(true);
393            }
394
395            // Zero rows: the INSERT was silently skipped by OR IGNORE.
396            // Only treat this as a dedup hit when a live note with the same
397            // non-empty external_id already exists in this namespace and kind.
398            // Any other ignored constraint (e.g. a PRIMARY KEY collision) must
399            // surface as an error rather than being misreported as a duplicate.
400            if let Some(ref ext_id) = ext_id_opt {
401                let is_dedup: bool = conn.query_row(
402                    "SELECT COUNT(*) > 0 FROM notes \
403                     WHERE namespace = ?1 \
404                       AND kind = ?2 \
405                       AND json_extract(properties, '$.external_id') = ?3 \
406                       AND deleted_at IS NULL",
407                    rusqlite::params![namespace, kind_str, ext_id],
408                    |row| row.get(0),
409                )?;
410                if is_dedup {
411                    return Ok(false);
412                }
413            }
414
415            // The INSERT was dropped for a reason other than an external_id
416            // collision.  Surface it as a constraint error.
417            Err(rusqlite::Error::SqliteFailure(
418                rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_CONSTRAINT),
419                Some(
420                    "try_insert_note: INSERT ignored for a constraint other than \
421                     external_id dedup; not masking as deduplication"
422                        .to_string(),
423                ),
424            ))
425        })
426        .await
427    }
428
429    async fn upsert_notes(&self, notes: Vec<Note>) -> Result<BatchWriteSummary, StorageError> {
430        let attempted = notes.len() as u64;
431
432        self.with_writer("upsert_notes", move |conn| {
433            conn.execute_batch("BEGIN IMMEDIATE")?;
434            let mut affected = 0u64;
435            let mut failed = 0u64;
436            let mut first_error = String::new();
437
438            for note in &notes {
439                let id_str = note.id.to_string();
440                let kind_str = note.kind.to_string();
441                let status_str = note.status.clone();
442                let properties_str = note
443                    .properties
444                    .as_ref()
445                    .map(|v| serde_json::to_string(v).unwrap_or_default());
446
447                match conn.execute(
448                    "INSERT OR REPLACE INTO notes \
449                     (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
450                      properties, created_at, updated_at, deleted_at) \
451                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
452                    rusqlite::params![
453                        id_str,
454                        &note.namespace,
455                        kind_str,
456                        status_str,
457                        &note.name,
458                        note.content,
459                        note.salience,
460                        note.decay_factor,
461                        note.expires_at,
462                        properties_str,
463                        note.created_at,
464                        note.updated_at,
465                        note.deleted_at,
466                    ],
467                ) {
468                    Ok(_) => affected += 1,
469                    Err(e) => {
470                        if first_error.is_empty() {
471                            first_error = e.to_string();
472                        }
473                        failed += 1;
474                    }
475                }
476            }
477
478            if let Err(e) = conn.execute_batch("COMMIT") {
479                let _ = conn.execute_batch("ROLLBACK");
480                return Err(e);
481            }
482            Ok(BatchWriteSummary {
483                attempted,
484                affected,
485                failed,
486                first_error,
487            })
488        })
489        .await
490    }
491
492    async fn get_note(&self, id: Uuid) -> Result<Option<Note>, StorageError> {
493        let id_str = id.to_string();
494
495        self.with_reader("get_note", move |conn| {
496            let mut stmt = conn.prepare(
497                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
498                 properties, created_at, updated_at, deleted_at \
499                 FROM notes WHERE id = ?1 AND deleted_at IS NULL",
500            )?;
501            let mut rows = stmt.query(rusqlite::params![id_str])?;
502            match rows.next()? {
503                Some(row) => Ok(Some(read_note(row)?)),
504                None => Ok(None),
505            }
506        })
507        .await
508    }
509
510    async fn get_note_including_deleted(&self, id: Uuid) -> Result<Option<Note>, StorageError> {
511        let id_str = id.to_string();
512
513        self.with_reader("get_note_including_deleted", move |conn| {
514            let mut stmt = conn.prepare(
515                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
516                 properties, created_at, updated_at, deleted_at \
517                 FROM notes WHERE id = ?1",
518            )?;
519            let mut rows = stmt.query(rusqlite::params![id_str])?;
520            match rows.next()? {
521                Some(row) => Ok(Some(read_note(row)?)),
522                None => Ok(None),
523            }
524        })
525        .await
526    }
527
528    async fn get_notes_batch(&self, ids: &[Uuid]) -> Result<Vec<Note>, StorageError> {
529        if ids.is_empty() {
530            return Ok(vec![]);
531        }
532        let id_strings: Vec<String> = ids.iter().map(|id| id.to_string()).collect();
533
534        self.with_reader("get_notes_batch", move |conn| {
535            let placeholders: String = (1..=id_strings.len())
536                .map(|i| format!("?{i}"))
537                .collect::<Vec<_>>()
538                .join(", ");
539            let sql = format!(
540                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
541                 properties, created_at, updated_at, deleted_at \
542                 FROM notes WHERE id IN ({placeholders}) AND deleted_at IS NULL"
543            );
544            let mut stmt = conn.prepare(&sql)?;
545            let params: Vec<&dyn rusqlite::types::ToSql> = id_strings
546                .iter()
547                .map(|s| s as &dyn rusqlite::types::ToSql)
548                .collect();
549            let rows = stmt.query_map(params.as_slice(), read_note)?;
550            let mut out = Vec::new();
551            for row in rows {
552                out.push(row?);
553            }
554            Ok(out)
555        })
556        .await
557    }
558
559    async fn delete_note(&self, id: Uuid, mode: DeleteMode) -> Result<bool, StorageError> {
560        let id_str = id.to_string();
561
562        match mode {
563            DeleteMode::Soft => {
564                self.with_writer("delete_note_soft", move |conn| {
565                    let now = chrono::Utc::now().timestamp_micros();
566                    let deleted = conn.execute(
567                        "UPDATE notes SET status = 'deleted', deleted_at = ?1 \
568                         WHERE id = ?2 AND deleted_at IS NULL",
569                        rusqlite::params![now, id_str],
570                    )?;
571                    Ok(deleted > 0)
572                })
573                .await
574            }
575            DeleteMode::Hard => {
576                self.with_writer("delete_note_hard", move |conn| {
577                    let deleted =
578                        conn.execute("DELETE FROM notes WHERE id = ?1", rusqlite::params![id_str])?;
579                    Ok(deleted > 0)
580                })
581                .await
582            }
583        }
584    }
585
586    async fn query_notes(
587        &self,
588        namespace: &str,
589        kind: Option<&str>,
590        page: PageRequest,
591    ) -> Result<Page<Note>, StorageError> {
592        let namespace = namespace.to_string();
593        let kind = kind.map(|k| k.to_string());
594
595        self.with_reader("query_notes", move |conn| {
596            let (count_sql, count_params) = build_note_where(&namespace, kind.as_deref());
597            let total: i64 = {
598                let sql = format!("SELECT COUNT(*) FROM notes{}", count_sql);
599                let mut stmt = conn.prepare(&sql)?;
600                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
601                    count_params.iter().map(|p| p.as_ref()).collect();
602                stmt.query_row(param_refs.as_slice(), |row| row.get(0))?
603            };
604
605            let (where_sql, mut data_params) = build_note_where(&namespace, kind.as_deref());
606            data_params.push(Box::new(page.limit as i64));
607            data_params.push(Box::new(page.offset as i64));
608
609            let limit_idx = data_params.len() - 1;
610            let offset_idx = data_params.len();
611
612            let data_sql = format!(
613                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
614                 properties, created_at, updated_at, deleted_at \
615                 FROM notes{} ORDER BY created_at DESC LIMIT ?{} OFFSET ?{}",
616                where_sql, limit_idx, offset_idx,
617            );
618
619            let mut stmt = conn.prepare(&data_sql)?;
620            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
621                data_params.iter().map(|p| p.as_ref()).collect();
622            let rows = stmt.query_map(param_refs.as_slice(), read_note)?;
623
624            let mut items = Vec::new();
625            for row in rows {
626                items.push(row?);
627            }
628
629            Ok(Page {
630                items,
631                total: Some(total as u64),
632            })
633        })
634        .await
635    }
636
637    async fn query_notes_filtered(
638        &self,
639        namespace: &str,
640        filter: &NoteFilter,
641        page: PageRequest,
642    ) -> Result<Page<Note>, StorageError> {
643        // Validate paths before entering spawn_blocking (closures return rusqlite::Error).
644        for pf in &filter.property_filters {
645            validate_json_path(&pf.json_path)?;
646        }
647        if let Some((path, _)) = &filter.order_by {
648            validate_json_path(path)?;
649        }
650
651        let namespace = namespace.to_string();
652        let filter = filter.clone();
653
654        self.with_reader("query_notes_filtered", move |conn| {
655            let (count_sql, count_params) = build_note_filter_where(&namespace, &filter)?;
656            let total: i64 = {
657                let sql = format!("SELECT COUNT(*) FROM notes{}", count_sql);
658                let mut stmt = conn.prepare(&sql)?;
659                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
660                    count_params.iter().map(|p| p.as_ref()).collect();
661                stmt.query_row(param_refs.as_slice(), |row| row.get(0))?
662            };
663
664            let (where_sql, mut data_params) = build_note_filter_where(&namespace, &filter)?;
665            data_params.push(Box::new(page.limit as i64));
666            data_params.push(Box::new(page.offset as i64));
667
668            let order_clause = match &filter.order_by {
669                Some((path, dir)) => {
670                    let dir_str = match dir {
671                        SortDir::Asc => "ASC",
672                        SortDir::Desc => "DESC",
673                    };
674                    format!(" ORDER BY {} {dir_str}", json_extract_expr(path))
675                }
676                None => " ORDER BY created_at DESC".to_string(),
677            };
678
679            let limit_idx = data_params.len() - 1;
680            let offset_idx = data_params.len();
681            let data_sql = format!(
682                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, \
683                 expires_at, properties, created_at, updated_at, deleted_at \
684                 FROM notes{}{order_clause} LIMIT ?{} OFFSET ?{}",
685                where_sql, limit_idx, offset_idx,
686            );
687
688            let mut stmt = conn.prepare(&data_sql)?;
689            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
690                data_params.iter().map(|p| p.as_ref()).collect();
691            let rows = stmt.query_map(param_refs.as_slice(), read_note)?;
692
693            let mut items = Vec::new();
694            for row in rows {
695                items.push(row?);
696            }
697
698            Ok(Page {
699                items,
700                total: Some(total as u64),
701            })
702        })
703        .await
704    }
705
706    async fn count_notes(&self, namespace: &str, kind: Option<&str>) -> Result<u64, StorageError> {
707        let namespace = namespace.to_string();
708        let kind = kind.map(|k| k.to_string());
709
710        self.with_reader("count_notes", move |conn| {
711            let (where_sql, params) = build_note_where(&namespace, kind.as_deref());
712            let sql = format!("SELECT COUNT(*) FROM notes{}", where_sql);
713            let mut stmt = conn.prepare(&sql)?;
714            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
715                params.iter().map(|p| p.as_ref()).collect();
716            let count: i64 = stmt.query_row(param_refs.as_slice(), |row| row.get(0))?;
717            Ok(count as u64)
718        })
719        .await
720    }
721}
722
723// =============================================================================
724// DDL
725// =============================================================================
726
727const NOTES_DDL: &str = include_str!("../../sql/notes-ddl.sql");
728
729pub(crate) fn ensure_notes_schema(conn: &rusqlite::Connection) -> Result<(), rusqlite::Error> {
730    conn.execute_batch(NOTES_DDL)
731}
732
733#[cfg(test)]
734#[path = "note_tests.rs"]
735mod tests;