khive-db 0.4.0

//! SQL-backed `NoteStore` implementation.

use std::sync::Arc;

use async_trait::async_trait;
use uuid::Uuid;

use khive_storage::error::StorageError;
use khive_storage::note::{FilterOp, Note, NoteFilter, SortDir};
use khive_storage::types::{
    BatchWriteSummary, DeleteMode, Page, PageRequest, SqlStatement, SqlValue,
};
use khive_storage::NoteStore;
use khive_storage::StorageCapability;

use crate::error::SqliteError;
use crate::pool::ConnectionPool;
use crate::sql_bridge::bind_params;
use crate::writer_task::WriterTaskHandle;

fn map_err(e: rusqlite::Error, op: &'static str) -> StorageError {
    StorageError::driver(StorageCapability::Notes, op, e)
}

fn map_sqlite_err(e: SqliteError, op: &'static str) -> StorageError {
    StorageError::driver(StorageCapability::Notes, op, e)
}

// ---------------------------------------------------------------------------
// Pure statement builders (ADR-099 B3 r6 structural cut) — see entity.rs's
// sibling block for the full rationale. `upsert_note`/`delete_note` below
// and ADR-099's atomic prepare path (`khive-runtime`) both call these.
// ---------------------------------------------------------------------------

/// The exact `INSERT OR REPLACE` this store's `upsert_note` issues.
pub fn note_upsert_statement(note: &Note) -> SqlStatement {
    let properties_str = note
        .properties
        .as_ref()
        .map(|v| serde_json::to_string(v).unwrap_or_default());
    SqlStatement {
        sql: "INSERT OR REPLACE INTO notes \
              (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
               properties, created_at, updated_at, deleted_at) \
              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)"
            .to_string(),
        params: vec![
            SqlValue::Text(note.id.to_string()),
            SqlValue::Text(note.namespace.clone()),
            SqlValue::Text(note.kind.to_string()),
            SqlValue::Text(note.status.clone()),
            match &note.name {
                Some(n) => SqlValue::Text(n.clone()),
                None => SqlValue::Null,
            },
            SqlValue::Text(note.content.clone()),
            match note.salience {
                Some(s) => SqlValue::Float(s),
                None => SqlValue::Null,
            },
            match note.decay_factor {
                Some(d) => SqlValue::Float(d),
                None => SqlValue::Null,
            },
            match note.expires_at {
                Some(e) => SqlValue::Integer(e),
                None => SqlValue::Null,
            },
            match properties_str {
                Some(p) => SqlValue::Text(p),
                None => SqlValue::Null,
            },
            SqlValue::Integer(note.created_at),
            SqlValue::Integer(note.updated_at),
            match note.deleted_at {
                Some(d) => SqlValue::Integer(d),
                None => SqlValue::Null,
            },
        ],
        label: Some("note-upsert".to_string()),
    }
}

/// The exact `properties`/`updated_at` `UPDATE` this store's
/// `update_note_properties` issues. A real `UPDATE` never triggers SQLite's
/// `INSERT OR REPLACE` delete+insert, so the row is patched in place (#780).
/// The `comm.probe` cursor is keyed on `notes_seq.seq`, which is fixed at
/// first insert and survives a delete+reinsert of the same note id, so this
/// is defensive rather than load-bearing for cursor correctness; a metadata
/// patch should never rewrite the row regardless.
pub fn note_update_properties_statement(
    id: Uuid,
    properties: &Option<serde_json::Value>,
    updated_at: i64,
) -> SqlStatement {
    let properties_str = properties
        .as_ref()
        .map(|v| serde_json::to_string(v).unwrap_or_default());
    SqlStatement {
        sql: "UPDATE notes SET properties = ?1, updated_at = ?2 \
              WHERE id = ?3 AND deleted_at IS NULL"
            .to_string(),
        params: vec![
            match properties_str {
                Some(p) => SqlValue::Text(p),
                None => SqlValue::Null,
            },
            SqlValue::Integer(updated_at),
            SqlValue::Text(id.to_string()),
        ],
        label: Some("note-update-properties".to_string()),
    }
}

/// The exact soft-delete `UPDATE` this store's `delete_note(Soft)` issues.
pub fn note_soft_delete_statement(id: Uuid, deleted_at: i64) -> SqlStatement {
    SqlStatement {
        sql: "UPDATE notes SET status = 'deleted', deleted_at = ?1 \
              WHERE id = ?2 AND deleted_at IS NULL"
            .to_string(),
        params: vec![
            SqlValue::Integer(deleted_at),
            SqlValue::Text(id.to_string()),
        ],
        label: Some("note-delete-soft".to_string()),
    }
}

/// The exact hard-delete `DELETE` this store's `delete_note(Hard)` issues.
pub fn note_hard_delete_statement(id: Uuid) -> SqlStatement {
    SqlStatement {
        sql: "DELETE FROM notes WHERE id = ?1".to_string(),
        params: vec![SqlValue::Text(id.to_string())],
        label: Some("note-delete-hard".to_string()),
    }
}

/// A NoteStore backed by SQLite. Namespace is the caller's responsibility.
///
/// UUID is globally unique — get/delete by ID alone. Query/count use the
/// namespace parameter as passed. The store is just a pool + is_file_backed.
pub struct SqlNoteStore {
    pool: Arc<ConnectionPool>,
    is_file_backed: bool,
    writer_task: Option<WriterTaskHandle>,
}

impl SqlNoteStore {
    /// Create a new store.
    pub fn new(pool: Arc<ConnectionPool>, is_file_backed: bool) -> Self {
        // Best-effort opt-in (ADR-067 Component A, mirrors entity.rs slice 1
        // policy): a missing writer task — flag off, spawn degraded, or no
        // Tokio runtime available at this first access — degrades to the
        // legacy pool-mutex path rather than failing construction.
        let writer_task = pool.writer_task_handle().ok().flatten();

        Self {
            pool,
            is_file_backed,
            writer_task,
        }
    }

    fn open_standalone_reader(&self) -> Result<rusqlite::Connection, StorageError> {
        let config = self.pool.config();
        let path = config.path.as_ref().ok_or_else(|| StorageError::Pool {
            operation: "note_reader".into(),
            message: "in-memory databases do not support standalone connections".into(),
        })?;

        let conn = rusqlite::Connection::open_with_flags(
            path,
            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY
                | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX
                | rusqlite::OpenFlags::SQLITE_OPEN_URI,
        )
        .map_err(|e| map_err(e, "open_note_reader"))?;

        conn.busy_timeout(config.busy_timeout)
            .map_err(|e| map_err(e, "open_note_reader"))?;
        conn.pragma_update(None, "foreign_keys", "ON")
            .map_err(|e| map_err(e, "open_note_reader"))?;
        conn.pragma_update(None, "synchronous", "NORMAL")
            .map_err(|e| map_err(e, "open_note_reader"))?;

        Ok(conn)
    }

    /// Route a single-row write through the pool-wide `WriterTask` when
    /// `KHIVE_WRITE_QUEUE=1` and a handle is available; otherwise fall back
    /// to the legacy pool-mutex path (ADR-067 Component A, Fork C slice 2).
    ///
    /// This is the routing point for single-statement `with_writer` callers
    /// in this store (`update_note_properties`, `delete_note`). `f` must be
    /// DML-only — on the flag-on path it runs inside the WriterTask's own
    /// transaction, so a bare `BEGIN IMMEDIATE` would violate SQLite's
    /// nested-transaction rule. `upsert_notes` (the batch method) does its
    /// own flag check and returns early on `Some`, so its fallback call
    /// into this helper only ever executes on the flag-off path
    /// (`self.writer_task` is `None` by construction whenever that call is
    /// reached) — no double-routing. Callers whose `f` issues more than one
    /// DML statement that must land atomically together (`upsert_note`,
    /// `try_insert_note`) use [`Self::with_writer_tx`] instead — see its doc
    /// comment (khive #827 Finding 2).
    async fn with_writer<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        if let Some(writer_task) = &self.writer_task {
            return writer_task
                .send(move |conn| f(conn).map_err(|e| map_err(e, op)))
                .await;
        }

        let pool = Arc::clone(&self.pool);
        tokio::task::spawn_blocking(move || {
            let guard = pool.try_writer().map_err(|e| map_sqlite_err(e, op))?;
            f(guard.conn()).map_err(|e| map_err(e, op))
        })
        .await
        .map_err(|e| StorageError::driver(StorageCapability::Notes, op, e))?
    }

    /// Like [`Self::with_writer`], but for callers whose closure issues more
    /// than one DML statement that must land atomically together (khive
    /// #827 Finding 2): a single-note insert immediately followed by
    /// `assign_note_seq`. On the flag-on path the WriterTask already wraps
    /// every request in its own `BEGIN IMMEDIATE`/`COMMIT`/`ROLLBACK`, so `f`
    /// is sent unwrapped, same as `with_writer`. On the flag-off (pool-mutex)
    /// path, `with_writer` runs `f` in SQLite's default autocommit mode --
    /// each statement inside `f` is its own implicit transaction -- so a
    /// crash or interleaving between the insert and the sequence assignment
    /// can strand a note that `comm.probe`'s `INNER JOIN notes_seq` will
    /// never see again. This wraps that path in one explicit transaction,
    /// matching `upsert_notes`' own flag-off branch.
    async fn with_writer_tx<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        if let Some(writer_task) = &self.writer_task {
            return writer_task
                .send(move |conn| f(conn).map_err(|e| map_err(e, op)))
                .await;
        }

        let pool = Arc::clone(&self.pool);
        tokio::task::spawn_blocking(move || {
            let guard = pool.try_writer().map_err(|e| map_sqlite_err(e, op))?;
            let conn = guard.conn();
            conn.execute_batch("BEGIN IMMEDIATE")
                .map_err(|e| map_err(e, op))?;

            match f(conn) {
                Ok(value) => match conn.execute_batch("COMMIT") {
                    Ok(()) => Ok(value),
                    Err(e) => {
                        let _ = conn.execute_batch("ROLLBACK");
                        Err(map_err(e, op))
                    }
                },
                Err(e) => {
                    let _ = conn.execute_batch("ROLLBACK");
                    Err(map_err(e, op))
                }
            }
        })
        .await
        .map_err(|e| StorageError::driver(StorageCapability::Notes, op, e))?
    }

    async fn with_reader<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        if self.is_file_backed {
            let conn = self.open_standalone_reader()?;
            tokio::task::spawn_blocking(move || f(&conn).map_err(|e| map_err(e, op)))
                .await
                .map_err(|e| StorageError::driver(StorageCapability::Notes, op, e))?
        } else {
            let pool = Arc::clone(&self.pool);
            tokio::task::spawn_blocking(move || {
                let guard = pool.reader().map_err(|e| map_sqlite_err(e, op))?;
                f(guard.conn()).map_err(|e| map_err(e, op))
            })
            .await
            .map_err(|e| StorageError::driver(StorageCapability::Notes, op, e))?
        }
    }
}

// =============================================================================
// Helpers
// =============================================================================

fn read_note(row: &rusqlite::Row<'_>) -> Result<Note, rusqlite::Error> {
    let id_str: String = row.get(0)?;
    let namespace: String = row.get(1)?;
    let kind: String = row.get(2)?;
    let status: String = row.get(3)?;
    let name: Option<String> = row.get(4)?;
    let content: String = row.get(5)?;
    let salience: Option<f64> = row.get(6)?;
    let decay_factor: Option<f64> = row.get(7)?;
    let expires_at: Option<i64> = row.get(8)?;
    let properties_str: Option<String> = row.get(9)?;
    let created_at: i64 = row.get(10)?;
    let updated_at: i64 = row.get(11)?;
    let deleted_at: Option<i64> = row.get(12)?;

    let id = parse_uuid(&id_str)?;

    let properties = properties_str
        .map(|s| {
            serde_json::from_str(&s).map_err(|e| {
                rusqlite::Error::FromSqlConversionFailure(
                    9,
                    rusqlite::types::Type::Text,
                    Box::new(e),
                )
            })
        })
        .transpose()?;

    Ok(Note {
        id,
        namespace,
        kind,
        status,
        name,
        content,
        salience,
        decay_factor,
        expires_at,
        properties,
        created_at,
        updated_at,
        deleted_at,
    })
}

fn parse_uuid(s: &str) -> Result<Uuid, rusqlite::Error> {
    Uuid::parse_str(s).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(e))
    })
}

/// DML-only batch upsert loop shared by both the legacy (flag-off) and
/// WriterTask-routed (flag-on) `upsert_notes` paths (ADR-067 Component A).
///
/// Issues no `BEGIN` / `COMMIT` / `ROLLBACK` itself — the caller owns the
/// enclosing transaction. Per-row failures are captured into
/// `BatchWriteSummary::failed`/`first_error` rather than aborting the loop,
/// matching the existing partial-success contract.
fn batch_upsert_notes(
    conn: &rusqlite::Connection,
    notes: &[Note],
    attempted: u64,
) -> Result<BatchWriteSummary, rusqlite::Error> {
    let mut affected = 0u64;
    let mut failed = 0u64;
    let mut first_error = String::new();

    for note in notes {
        let id_str = note.id.to_string();
        let kind_str = note.kind.to_string();
        let status_str = note.status.clone();
        let properties_str = note
            .properties
            .as_ref()
            .map(|v| serde_json::to_string(v).unwrap_or_default());

        match conn.execute(
            "INSERT OR REPLACE INTO notes \
             (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
              properties, created_at, updated_at, deleted_at) \
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
            rusqlite::params![
                id_str,
                &note.namespace,
                kind_str,
                status_str,
                &note.name,
                note.content,
                note.salience,
                note.decay_factor,
                note.expires_at,
                properties_str,
                note.created_at,
                note.updated_at,
                note.deleted_at,
            ],
        ) {
            Ok(_) => {
                assign_note_seq(conn, &id_str)?;
                affected += 1;
            }
            Err(e) => {
                if first_error.is_empty() {
                    first_error = e.to_string();
                }
                failed += 1;
            }
        }
    }

    Ok(BatchWriteSummary {
        attempted,
        affected,
        failed,
        first_error,
    })
}

/// Assign a note id its durable, non-reusing sequence number the first time
/// it is inserted (khive #827 — see `sql/007-notes-seq.sql`). `INSERT OR
/// IGNORE` makes this idempotent across an `INSERT OR REPLACE`
/// delete+reinsert of the same note id: the sequence value is fixed at the
/// note's first insert and never reassigned, unlike `notes`' own implicit
/// rowid.
fn assign_note_seq(conn: &rusqlite::Connection, note_id: &str) -> Result<(), rusqlite::Error> {
    conn.execute(
        "INSERT OR IGNORE INTO notes_seq (note_id) VALUES (?1)",
        rusqlite::params![note_id],
    )?;
    Ok(())
}

fn build_note_where(
    namespace: &str,
    kind: Option<&str>,
) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
    let mut conditions: Vec<String> = vec![
        "namespace = ?1".to_string(),
        "deleted_at IS NULL".to_string(),
    ];
    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(namespace.to_string())];

    if let Some(k) = kind {
        params.push(Box::new(k.to_string()));
        conditions.push(format!("kind = ?{}", params.len()));
    }

    let clause = format!(" WHERE {}", conditions.join(" AND "));
    (clause, params)
}

/// Validate that a json_path is safe to interpolate into SQL.
/// Accepts only `$.field` or `$.field.subfield` paths with alphanumeric/underscore segments.
fn validate_json_path(path: &str) -> Result<(), StorageError> {
    let valid = path.starts_with("$.")
        && path[2..].split('.').all(|part| {
            !part.is_empty() && part.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
        });
    if valid {
        Ok(())
    } else {
        Err(StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "query_notes_filtered".into(),
            message: format!("invalid JSON path for note filter: {path:?}"),
        })
    }
}

fn json_extract_expr(path: &str) -> String {
    format!("json_extract(properties, '{path}')")
}

fn json_type_expr(path: &str) -> String {
    format!("json_type(properties, '{path}')")
}

fn sql_value_param(value: &SqlValue) -> Result<Box<dyn rusqlite::types::ToSql>, rusqlite::Error> {
    Ok(match value {
        SqlValue::Null => Box::new(Option::<String>::None),
        SqlValue::Bool(v) => Box::new(*v as i64),
        SqlValue::Integer(v) => Box::new(*v),
        SqlValue::Float(v) => Box::new(*v),
        SqlValue::Text(v) => Box::new(v.clone()),
        SqlValue::Blob(v) => Box::new(v.clone()),
        SqlValue::Json(v) => Box::new(
            serde_json::to_string(v)
                .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
        ),
        SqlValue::Uuid(v) => Box::new(v.to_string()),
        SqlValue::Timestamp(v) => Box::new(v.timestamp_micros()),
    })
}

fn build_note_filter_where(
    namespace: &str,
    filter: &NoteFilter,
) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), rusqlite::Error> {
    // When filter.namespaces is non-empty use `namespace IN (...)` for
    // multi-namespace read visibility. Otherwise fall back to equality.
    let (ns_condition, ns_params): (String, Vec<Box<dyn rusqlite::types::ToSql>>) =
        if !filter.namespaces.is_empty() {
            let placeholders: Vec<String> = (1..=filter.namespaces.len())
                .map(|i| format!("?{i}"))
                .collect();
            let params: Vec<Box<dyn rusqlite::types::ToSql>> = filter
                .namespaces
                .iter()
                .map(|ns| -> Box<dyn rusqlite::types::ToSql> { Box::new(ns.clone()) })
                .collect();
            (
                format!("namespace IN ({})", placeholders.join(", ")),
                params,
            )
        } else {
            (
                "namespace = ?1".to_string(),
                vec![Box::new(namespace.to_string())],
            )
        };

    let mut conditions = vec![ns_condition, "deleted_at IS NULL".to_string()];
    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = ns_params;

    if let Some(kind) = &filter.kind {
        params.push(Box::new(kind.clone()));
        conditions.push(format!("kind = ?{}", params.len()));
    }

    for pf in &filter.property_filters {
        match &pf.op {
            FilterOp::EqOrMissing => {
                let expr = json_extract_expr(&pf.json_path);
                params.push(sql_value_param(&pf.value)?);
                conditions.push(format!(
                    "({expr} = ?{n} OR {expr} IS NULL)",
                    n = params.len()
                ));
            }
            FilterOp::JsonTypeEq => {
                let type_expr = json_type_expr(&pf.json_path);
                params.push(sql_value_param(&pf.value)?);
                conditions.push(format!("{type_expr} = ?{}", params.len()));
            }
            FilterOp::JsonTypeNeMissing => {
                let type_expr = json_type_expr(&pf.json_path);
                params.push(sql_value_param(&pf.value)?);
                let n = params.len();
                conditions.push(format!("({type_expr} IS NULL OR {type_expr} != ?{n})"));
            }
            FilterOp::In(values) => {
                let expr = json_extract_expr(&pf.json_path);
                if values.is_empty() {
                    // An empty set can never match any row.
                    conditions.push("0".to_string());
                    continue;
                }
                let mut placeholders = Vec::with_capacity(values.len());
                for v in values {
                    params.push(sql_value_param(v)?);
                    placeholders.push(format!("?{}", params.len()));
                }
                conditions.push(format!("{expr} IN ({})", placeholders.join(", ")));
            }
            FilterOp::NotInOrMissing(values) => {
                let expr = json_extract_expr(&pf.json_path);
                if values.is_empty() {
                    // Nothing to exclude — every row (including missing) matches.
                    continue;
                }
                let mut placeholders = Vec::with_capacity(values.len());
                for v in values {
                    params.push(sql_value_param(v)?);
                    placeholders.push(format!("?{}", params.len()));
                }
                conditions.push(format!(
                    "({expr} IS NULL OR {expr} NOT IN ({}))",
                    placeholders.join(", ")
                ));
            }
            _ => {
                let expr = json_extract_expr(&pf.json_path);
                let op = match pf.op {
                    FilterOp::Eq => "=",
                    FilterOp::Ne => "!=",
                    FilterOp::Lt => "<",
                    FilterOp::Lte => "<=",
                    FilterOp::Gt => ">",
                    FilterOp::Gte => ">=",
                    FilterOp::EqOrMissing
                    | FilterOp::JsonTypeEq
                    | FilterOp::JsonTypeNeMissing
                    | FilterOp::In(_)
                    | FilterOp::NotInOrMissing(_) => {
                        unreachable!()
                    }
                };
                params.push(sql_value_param(&pf.value)?);
                conditions.push(format!("{expr} {op} ?{}", params.len()));
            }
        }
    }

    if let Some(min_ts) = filter.min_created_at {
        params.push(Box::new(min_ts));
        conditions.push(format!("created_at >= ?{}", params.len()));
    }

    Ok((format!(" WHERE {}", conditions.join(" AND ")), params))
}

// =============================================================================
// NoteStore implementation
// =============================================================================

#[async_trait]
impl NoteStore for SqlNoteStore {
    async fn upsert_note(&self, note: Note) -> Result<(), StorageError> {
        let id_str = note.id.to_string();
        let statement = note_upsert_statement(&note);
        self.with_writer_tx("upsert_note", move |conn| {
            let mut stmt = conn.prepare(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            stmt.raw_execute()?;
            assign_note_seq(conn, &id_str)?;
            Ok(())
        })
        .await
    }

    async fn update_note_properties(
        &self,
        id: Uuid,
        properties: Option<serde_json::Value>,
        updated_at: i64,
    ) -> Result<bool, StorageError> {
        let statement = note_update_properties_statement(id, &properties, updated_at);
        self.with_writer("update_note_properties", move |conn| {
            let mut stmt = conn.prepare(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            Ok(stmt.raw_execute()? > 0)
        })
        .await
    }

    async fn try_insert_note(&self, note: Note) -> Result<bool, StorageError> {
        let namespace = note.namespace.clone();
        let id_str = note.id.to_string();
        let kind_str = note.kind.to_string();
        let status_str = note.status.clone();
        let properties_str = note
            .properties
            .as_ref()
            .map(|v| serde_json::to_string(v).unwrap_or_default());

        // Extract external_id (if any) for dedup verification after a zero-row insert.
        let ext_id_opt: Option<String> = note
            .properties
            .as_ref()
            .and_then(|v| v.get("external_id"))
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .map(|s| s.to_string());

        self.with_writer_tx("try_insert_note", move |conn| {
            let rows = conn.execute(
                "INSERT OR IGNORE INTO notes \
                 (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                  properties, created_at, updated_at, deleted_at) \
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
                rusqlite::params![
                    id_str,
                    namespace,
                    kind_str,
                    status_str,
                    note.name,
                    note.content,
                    note.salience,
                    note.decay_factor,
                    note.expires_at,
                    properties_str,
                    note.created_at,
                    note.updated_at,
                    note.deleted_at,
                ],
            )?;

            if rows > 0 {
                assign_note_seq(conn, &id_str)?;
                return Ok(true);
            }

            // Zero rows: the INSERT was silently skipped by OR IGNORE.
            // Only treat this as a dedup hit when a live note with the same
            // non-empty external_id already exists in this namespace and kind.
            // Any other ignored constraint (e.g. a PRIMARY KEY collision) must
            // surface as an error rather than being misreported as a duplicate.
            if let Some(ref ext_id) = ext_id_opt {
                let is_dedup: bool = conn.query_row(
                    "SELECT COUNT(*) > 0 FROM notes \
                     WHERE namespace = ?1 \
                       AND kind = ?2 \
                       AND json_extract(properties, '$.external_id') = ?3 \
                       AND deleted_at IS NULL",
                    rusqlite::params![namespace, kind_str, ext_id],
                    |row| row.get(0),
                )?;
                if is_dedup {
                    return Ok(false);
                }
            }

            // The INSERT was dropped for a reason other than an external_id
            // collision.  Surface it as a constraint error.
            Err(rusqlite::Error::SqliteFailure(
                rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_CONSTRAINT),
                Some(
                    "try_insert_note: INSERT ignored for a constraint other than \
                     external_id dedup; not masking as deduplication"
                        .to_string(),
                ),
            ))
        })
        .await
    }

    async fn upsert_notes(&self, notes: Vec<Note>) -> Result<BatchWriteSummary, StorageError> {
        let attempted = notes.len() as u64;

        // khive #827 Finding 2 (round 3): route through `with_writer_tx`
        // instead of hand-rolling BEGIN IMMEDIATE/COMMIT/ROLLBACK here. The
        // old flag-off path only rolled back when the final COMMIT failed —
        // an earlier error from `batch_upsert_notes` (e.g. a failed
        // `assign_note_seq`) propagated via `?` straight out of the closure,
        // skipping ROLLBACK entirely and leaving BEGIN IMMEDIATE open on the
        // shared pool-mutex connection, poisoning every later write on that
        // connection. `with_writer_tx` rolls back on ANY error from `f`, on
        // both the flag-on (WriterTask, which wraps its own transaction) and
        // flag-off (pool-mutex) paths.
        self.with_writer_tx("upsert_notes", move |conn| {
            let _tx_handle =
                khive_storage::tx_registry::register(Some("note_upsert_batch".to_string()));
            batch_upsert_notes(conn, &notes, attempted)
        })
        .await
    }

    async fn get_note(&self, id: Uuid) -> Result<Option<Note>, StorageError> {
        let id_str = id.to_string();

        self.with_reader("get_note", move |conn| {
            let mut stmt = conn.prepare(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                 properties, created_at, updated_at, deleted_at \
                 FROM notes WHERE id = ?1 AND deleted_at IS NULL",
            )?;
            let mut rows = stmt.query(rusqlite::params![id_str])?;
            match rows.next()? {
                Some(row) => Ok(Some(read_note(row)?)),
                None => Ok(None),
            }
        })
        .await
    }

    async fn get_note_including_deleted(&self, id: Uuid) -> Result<Option<Note>, StorageError> {
        let id_str = id.to_string();

        self.with_reader("get_note_including_deleted", move |conn| {
            let mut stmt = conn.prepare(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                 properties, created_at, updated_at, deleted_at \
                 FROM notes WHERE id = ?1",
            )?;
            let mut rows = stmt.query(rusqlite::params![id_str])?;
            match rows.next()? {
                Some(row) => Ok(Some(read_note(row)?)),
                None => Ok(None),
            }
        })
        .await
    }

    async fn get_notes_batch(&self, ids: &[Uuid]) -> Result<Vec<Note>, StorageError> {
        if ids.is_empty() {
            return Ok(vec![]);
        }
        let id_strings: Vec<String> = ids.iter().map(|id| id.to_string()).collect();

        self.with_reader("get_notes_batch", move |conn| {
            let placeholders: String = (1..=id_strings.len())
                .map(|i| format!("?{i}"))
                .collect::<Vec<_>>()
                .join(", ");
            let sql = format!(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                 properties, created_at, updated_at, deleted_at \
                 FROM notes WHERE id IN ({placeholders}) AND deleted_at IS NULL"
            );
            let mut stmt = conn.prepare(&sql)?;
            let params: Vec<&dyn rusqlite::types::ToSql> = id_strings
                .iter()
                .map(|s| s as &dyn rusqlite::types::ToSql)
                .collect();
            let rows = stmt.query_map(params.as_slice(), read_note)?;
            let mut out = Vec::new();
            for row in rows {
                out.push(row?);
            }
            Ok(out)
        })
        .await
    }

    async fn delete_note(&self, id: Uuid, mode: DeleteMode) -> Result<bool, StorageError> {
        match mode {
            DeleteMode::Soft => {
                let now = chrono::Utc::now().timestamp_micros();
                let statement = note_soft_delete_statement(id, now);
                self.with_writer("delete_note_soft", move |conn| {
                    let mut stmt = conn.prepare(&statement.sql)?;
                    bind_params(&mut stmt, &statement.params)?;
                    Ok(stmt.raw_execute()? > 0)
                })
                .await
            }
            DeleteMode::Hard => {
                let statement = note_hard_delete_statement(id);
                self.with_writer("delete_note_hard", move |conn| {
                    let mut stmt = conn.prepare(&statement.sql)?;
                    bind_params(&mut stmt, &statement.params)?;
                    Ok(stmt.raw_execute()? > 0)
                })
                .await
            }
        }
    }

    async fn query_notes(
        &self,
        namespace: &str,
        kind: Option<&str>,
        page: PageRequest,
    ) -> Result<Page<Note>, StorageError> {
        let namespace = namespace.to_string();
        let kind = kind.map(|k| k.to_string());
        let limit_i64 = i64::from(page.limit);
        let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "query_notes".into(),
            message: format!(
                "PageRequest: offset must be <= i64::MAX, got {}",
                page.offset
            ),
        })?;

        self.with_reader("query_notes", move |conn| {
            let (count_sql, count_params) = build_note_where(&namespace, kind.as_deref());
            let total: i64 = {
                let sql = format!("SELECT COUNT(*) FROM notes{}", count_sql);
                let mut stmt = conn.prepare(&sql)?;
                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                    count_params.iter().map(|p| p.as_ref()).collect();
                stmt.query_row(param_refs.as_slice(), |row| row.get(0))?
            };

            let (where_sql, mut data_params) = build_note_where(&namespace, kind.as_deref());
            data_params.push(Box::new(limit_i64));
            data_params.push(Box::new(offset_i64));

            let limit_idx = data_params.len() - 1;
            let offset_idx = data_params.len();

            let data_sql = format!(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                 properties, created_at, updated_at, deleted_at \
                 FROM notes{} ORDER BY created_at DESC LIMIT ?{} OFFSET ?{}",
                where_sql, limit_idx, offset_idx,
            );

            let mut stmt = conn.prepare(&data_sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                data_params.iter().map(|p| p.as_ref()).collect();
            let rows = stmt.query_map(param_refs.as_slice(), read_note)?;

            let mut items = Vec::new();
            for row in rows {
                items.push(row?);
            }

            Ok(Page {
                items,
                total: Some(total as u64),
            })
        })
        .await
    }

    async fn query_notes_filtered(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        page: PageRequest,
    ) -> Result<Page<Note>, StorageError> {
        // Validate paths before entering spawn_blocking (closures return rusqlite::Error).
        for pf in &filter.property_filters {
            validate_json_path(&pf.json_path)?;
        }
        if let Some((path, _)) = &filter.order_by {
            validate_json_path(path)?;
        }

        let namespace = namespace.to_string();
        let filter = filter.clone();
        let limit_i64 = i64::from(page.limit);
        let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "query_notes_filtered".into(),
            message: format!(
                "PageRequest: offset must be <= i64::MAX, got {}",
                page.offset
            ),
        })?;

        self.with_reader("query_notes_filtered", move |conn| {
            let (count_sql, count_params) = build_note_filter_where(&namespace, &filter)?;
            let total: i64 = {
                let sql = format!("SELECT COUNT(*) FROM notes{}", count_sql);
                let mut stmt = conn.prepare(&sql)?;
                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                    count_params.iter().map(|p| p.as_ref()).collect();
                stmt.query_row(param_refs.as_slice(), |row| row.get(0))?
            };

            let (where_sql, mut data_params) = build_note_filter_where(&namespace, &filter)?;
            data_params.push(Box::new(limit_i64));
            data_params.push(Box::new(offset_i64));

            let order_clause = match &filter.order_by {
                Some((path, dir)) => {
                    let dir_str = match dir {
                        SortDir::Asc => "ASC",
                        SortDir::Desc => "DESC",
                    };
                    format!(" ORDER BY {} {dir_str}", json_extract_expr(path))
                }
                None => " ORDER BY created_at DESC".to_string(),
            };

            let limit_idx = data_params.len() - 1;
            let offset_idx = data_params.len();
            let data_sql = format!(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, \
                 expires_at, properties, created_at, updated_at, deleted_at \
                 FROM notes{}{order_clause} LIMIT ?{} OFFSET ?{}",
                where_sql, limit_idx, offset_idx,
            );

            let mut stmt = conn.prepare(&data_sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                data_params.iter().map(|p| p.as_ref()).collect();
            let rows = stmt.query_map(param_refs.as_slice(), read_note)?;

            let mut items = Vec::new();
            for row in rows {
                items.push(row?);
            }

            Ok(Page {
                items,
                total: Some(total as u64),
            })
        })
        .await
    }

    async fn query_notes_filtered_bounded(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        max_rows: u32,
    ) -> Result<Vec<Note>, StorageError> {
        for pf in &filter.property_filters {
            validate_json_path(&pf.json_path)?;
        }
        if let Some((path, _)) = &filter.order_by {
            validate_json_path(path)?;
        }

        let namespace = namespace.to_string();
        let filter = filter.clone();
        let limit_i64 = i64::from(max_rows) + 1;

        self.with_reader("query_notes_filtered_bounded", move |conn| {
            let (where_sql, mut data_params) = build_note_filter_where(&namespace, &filter)?;
            data_params.push(Box::new(limit_i64));
            let limit_idx = data_params.len();

            // Tie-break on `id` in addition to the primary sort key so the
            // snapshot ordering is fully deterministic even when many rows
            // share the same `created_at` (or the same custom sort value).
            let order_clause = match &filter.order_by {
                Some((path, dir)) => {
                    let dir_str = match dir {
                        SortDir::Asc => "ASC",
                        SortDir::Desc => "DESC",
                    };
                    format!(" ORDER BY {} {dir_str}, id ASC", json_extract_expr(path))
                }
                None => " ORDER BY created_at DESC, id ASC".to_string(),
            };

            let data_sql = format!(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, \
                 expires_at, properties, created_at, updated_at, deleted_at \
                 FROM notes{where_sql}{order_clause} LIMIT ?{limit_idx}",
            );

            let mut stmt = conn.prepare(&data_sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                data_params.iter().map(|p| p.as_ref()).collect();
            let rows = stmt.query_map(param_refs.as_slice(), read_note)?;

            let mut items = Vec::new();
            for row in rows {
                items.push(row?);
            }
            Ok(items)
        })
        .await
    }

    async fn count_notes(&self, namespace: &str, kind: Option<&str>) -> Result<u64, StorageError> {
        let namespace = namespace.to_string();
        let kind = kind.map(|k| k.to_string());

        self.with_reader("count_notes", move |conn| {
            let (where_sql, params) = build_note_where(&namespace, kind.as_deref());
            let sql = format!("SELECT COUNT(*) FROM notes{}", where_sql);
            let mut stmt = conn.prepare(&sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                params.iter().map(|p| p.as_ref()).collect();
            let count: i64 = stmt.query_row(param_refs.as_slice(), |row| row.get(0))?;
            Ok(count as u64)
        })
        .await
    }
}

// =============================================================================
// DDL
// =============================================================================

const NOTES_DDL: &str = include_str!("../../sql/notes-ddl.sql");

/// Same anti-join repair as `sql/008-notes-seq-repair.sql` (the V8 forward
/// migration) -- shared via `include_str!` from that single source file
/// rather than duplicated as SQL text. `INSERT OR IGNORE` targets notes
/// still missing a `notes_seq` row specifically, so it is correct to run
/// against a fresh ledger, a partially populated one, or an already fully
/// repaired one.
const NOTES_SEQ_REPAIR_DDL: &str = include_str!("../../sql/008-notes-seq-repair.sql");

pub(crate) fn ensure_notes_schema(conn: &rusqlite::Connection) -> Result<(), rusqlite::Error> {
    conn.execute_batch(NOTES_DDL)
}

/// Anti-join backfill of `notes_seq` for any note still missing a row
/// (khive #827 round 3). Scans `notes` in full, so callers MUST gate this to
/// run at most once per backend/pool rather than on every store acquisition
/// (khive #827 round 4 perf finding) -- see
/// `StorageBackend::notes_for_namespace`.
pub(crate) fn repair_notes_seq(conn: &rusqlite::Connection) -> Result<(), rusqlite::Error> {
    conn.execute_batch(NOTES_SEQ_REPAIR_DDL)
}

#[cfg(test)]
#[path = "note_tests.rs"]
mod tests;