khive_storage/

note.rs

//! Note storage capability — temporal-referential record CRUD.

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::BTreeMap;
use uuid::Uuid;

use crate::types::{
    BatchWriteSummary, BoundedCount, DeleteMode, Page, PageRequest, SeekCursor, SeekPage, SqlValue,
    StorageResult,
};

/// A storage-level note record. Flat, SQL-friendly representation.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Note {
    pub id: Uuid,
    pub namespace: String,
    pub kind: String,
    pub status: String,
    pub name: Option<String>,
    pub content: String,
    pub salience: Option<f64>,
    pub decay_factor: Option<f64>,
    pub expires_at: Option<i64>,
    pub properties: Option<Value>,
    pub created_at: i64,
    pub updated_at: i64,
    pub deleted_at: Option<i64>,
    /// Immutable caller-chosen identity, unique among live notes of the same namespace and kind.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub key: Option<String>,
    /// Persisted revision, assigned by storage and advanced on every matched update.
    #[serde(default = "initial_note_version")]
    pub version: i64,
}

/// The columns needed to decide whether a note may appear in an ID-based result.
/// Includes soft-deleted rows so the caller can apply its visibility policy.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct NoteVisibility {
    pub id: Uuid,
    pub namespace: String,
    pub deleted_at: Option<i64>,
}

const fn initial_note_version() -> i64 {
    1
}

impl Note {
    /// Create a new note with a generated UUID and current timestamp.
    pub fn new(
        namespace: impl Into<String>,
        kind: impl Into<String>,
        content: impl Into<String>,
    ) -> Self {
        let now = chrono::Utc::now().timestamp_micros();
        Self {
            id: Uuid::new_v4(),
            namespace: namespace.into(),
            kind: kind.into(),
            status: "active".to_string(),
            name: None,
            content: content.into(),
            salience: None,
            decay_factor: None,
            expires_at: None,
            properties: None,
            created_at: now,
            updated_at: now,
            deleted_at: None,
            key: None,
            version: 1,
        }
    }

    /// Set the note display name.
    pub fn with_name(mut self, n: impl Into<String>) -> Self {
        self.name = Some(n.into());
        self
    }

    /// Set salience (infallible). Rejects non-finite values by returning `self`
    /// unchanged; clamps finite values to `[0.0, 1.0]`. Prefer
    /// [`try_with_salience`](Self::try_with_salience) at public boundaries.
    pub fn with_salience(mut self, s: f64) -> Self {
        if !s.is_finite() {
            return self;
        }
        self.salience = Some(s.clamp(0.0, 1.0));
        self
    }

    /// Set decay factor (infallible). Rejects non-finite values by returning
    /// `self` unchanged; floors finite values at `0.0`. Prefer
    /// [`try_with_decay`](Self::try_with_decay) at public boundaries.
    pub fn with_decay(mut self, d: f64) -> Self {
        if !d.is_finite() {
            return self;
        }
        self.decay_factor = Some(d.max(0.0));
        self
    }

    /// Set salience with validation. Returns an error for non-finite or
    /// out-of-range `[0.0, 1.0]` values.
    pub fn try_with_salience(mut self, s: f64) -> Result<Self, String> {
        if !s.is_finite() {
            return Err(format!("salience must be finite, got {s}"));
        }
        if !(0.0..=1.0).contains(&s) {
            return Err(format!("salience must be in [0.0, 1.0], got {s}"));
        }
        self.salience = Some(s);
        Ok(self)
    }

    /// Set decay factor with validation. Returns an error for non-finite or
    /// negative values.
    pub fn try_with_decay(mut self, d: f64) -> Result<Self, String> {
        if !d.is_finite() {
            return Err(format!("decay_factor must be finite, got {d}"));
        }
        if d < 0.0 {
            return Err(format!("decay_factor must be >= 0.0, got {d}"));
        }
        self.decay_factor = Some(d);
        Ok(self)
    }

    /// Set the note properties JSON blob.
    pub fn with_properties(mut self, p: Value) -> Self {
        self.properties = Some(p);
        self
    }
}

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

/// Sort direction for filtered note queries.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SortDir {
    Asc,
    Desc,
}

/// Comparison operator for a [`PropertyFilter`] on a JSON path.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FilterOp {
    Eq,
    /// Matches rows where the JSON field equals the value OR the field is absent/NULL.
    /// Used for properties that may be missing in legacy rows (e.g. `$.read`).
    EqOrMissing,
    /// Matches the supplied value using the indexable
    /// `ifnull(json_extract(...), '')` expression. The value may be any
    /// [`SqlValue`] accepted by the SQL adapter; in particular, `Text("")`
    /// matches both missing values and present-but-empty values.
    EqOrMissingIndexed,
    /// Matches rows where the JSON field is absent or SQL-NULL.
    JsonTypeMissing,
    /// Matches a missing path, JSON null, or JSON text empty after SQLite's
    /// ASCII-SPACE-only `trim`. Tabs and Unicode whitespace do not match.
    /// `PropertyFilter.value` is unused.
    MissingNullOrSpaceEmptyText,
    /// Matches JSON boolean true or the exact JSON text "true", not numeric 1.
    /// `PropertyFilter.value` is unused.
    TrueOrTextTrue,
    /// Matches rows where the JSON field is absent or explicitly JSON `null`,
    /// while constraining its index key to the empty recipient key. This is
    /// the index-friendly legacy-recipient partition used with
    /// [`FilterOp::EqOrMissingIndexed`].
    JsonTypeMissingOrNullIndexed,
    /// Combines the exact-value and legacy-recipient partitions
    /// (`EqOrMissingIndexed` + `JsonTypeMissingOrNullIndexed`) into one
    /// predicate over the same indexable `ifnull(json_extract(...), '')`
    /// expression, so a single index seek serves both partitions instead of
    /// two separate bounded queries. Matches rows where the field equals the
    /// value, OR the field is absent/JSON-`null`. A present-but-empty JSON
    /// string value does NOT match through the legacy branch — the same
    /// `json_type` guard `JsonTypeMissingOrNullIndexed` uses excludes it —
    /// so this reproduces `EqOrMissing` exactly, given a non-empty compared
    /// value.
    EqOrLegacyIndexed,
    /// Matches rows where a JSON text field equals the value, while treating
    /// every missing or non-text value as that same value. The SQL adapter
    /// emits `CASE WHEN json_type(...) = 'text' THEN json_extract(...) ELSE
    /// value END = value`, mirroring callers whose read model assigns one
    /// textual default to absent, JSON-null, and malformed legacy values.
    TextEqOrNonText,
    /// Matches a JSON text field against the supplied set, or any missing or
    /// non-text value. The non-text branch still matches when the set is empty.
    /// `PropertyFilter.value` is unused; the set lives in this variant.
    TextInOrNonText(Vec<SqlValue>),
    Ne,
    Lt,
    Lte,
    Gt,
    Gte,
    /// Matches rows where `json_type(properties, path) = value`.
    /// Value must be a SQLite json_type string literal: 'true', 'false', 'integer',
    /// 'real', 'text', 'array', 'object', or 'null'.
    JsonTypeEq,
    /// Matches rows where the json_type is absent (NULL) OR differs from value.
    /// Equivalent to `json_type IS NULL OR json_type != value`.
    /// Value must be a SQLite json_type string literal: 'true', 'false',
    /// 'integer', 'real', 'text', 'array', 'object', or 'null'. Used for
    /// unread filter: matches any `$.read` that is NOT the JSON boolean true.
    JsonTypeNeMissing,
    /// Matches rows where `json_extract(properties, path)` equals any value in
    /// the set. A row with a missing/NULL property does not match — use
    /// `TextInOrNonText` when a textual read model includes missing/non-text
    /// legacy values. `PropertyFilter.value` is unused for this op; the set
    /// lives in the variant itself.
    In(Vec<SqlValue>),
    /// Matches rows where the property is missing/NULL OR its value is not in
    /// the set. Used to exclude a closed set while treating an unset property
    /// as included (e.g. comm inbox excludes `outbound`). `PropertyFilter.value` is
    /// unused for this op; the set lives in the variant itself.
    NotInOrMissing(Vec<SqlValue>),
    /// Matches rows whose JSON text field starts with the supplied prefix.
    /// Rendered as an index-seekable half-open range over the plain
    /// `json_extract` expression (`expr >= prefix AND expr < next(prefix)`,
    /// where `next` increments the prefix's last code point), so an index
    /// keyed on that expression serves the seek instead of a scan. SQLite
    /// sorts non-text values outside the text range, so a missing, null or
    /// numeric field never matches. `PropertyFilter.value` is the prefix and
    /// must be `SqlValue::Text`; an empty prefix matches every text value.
    TextStartsWithIndexed,
    /// Match a recipient's first colon-terminated channel prefix by equality
    /// on its indexed bucket. The value must be a nonempty `SqlValue::Text`
    /// prefix with exactly one trailing colon, such as `email:`. Arbitrary
    /// partial prefixes use `TextStartsWithIndexed` instead.
    TextColonPrefixBucketIndexed,
    /// Keep only RFC 3339 text values that parse as UTC instants.
    Rfc3339Valid,
    /// Compare parsed UTC instants, including subsecond precision and offsets.
    /// The value is a `Timestamp` or RFC 3339 `Text`.
    Rfc3339Gte,
    Rfc3339Lte,
    /// Matches an RFC 3339 instant at or before the supplied timestamp, or a
    /// missing, null, non-text, or malformed value. The outbox uses this to
    /// fail open on legacy retry deadlines without filtering after LIMIT.
    Rfc3339LteOrInvalid,
}

/// A single `json_extract(properties, '$.field') op value` predicate.
///
/// Callers import this as `khive_storage::note::PropertyFilter` to avoid
/// collision with the vector-metadata `PropertyFilter` in `khive_storage::types`.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PropertyFilter {
    pub json_path: String,
    pub op: FilterOp,
    pub value: SqlValue,
}

/// A predicate on one literal top-level property key, rechecked at write time.
/// Explicit JSON null is present and never matches an absent-key branch.
#[derive(Clone, Debug)]
pub enum NotePropertyPrecondition {
    /// `json_extract(properties, path) = value`, with the ordinary [`SqlValue`]
    /// binding rules. SQL NULL equality does not match; this is not typed JSON equality.
    ExtractEquals { key: String, value: SqlValue },
    /// An absent key (`json_type IS NULL`) or [`Self::ExtractEquals`].
    AbsentOrExtractEquals { key: String, value: SqlValue },
    /// An absent key, or a JSON text value exactly equal to `value`.
    AbsentOrTextEquals { key: String, value: String },
    /// JSON boolean `true` or the exact JSON string `"true"`; numeric 1 is excluded.
    TrueOrTextTrue { key: String },
}

/// A bounded, atomic partial update of a note's properties and retention metadata.
#[derive(Clone, Debug)]
pub struct NotePropertyPatch {
    /// AND-combined predicates, at most 32. Keys are literal top-level keys.
    pub preconditions: Vec<NotePropertyPrecondition>,
    /// Between 1 and 32 literal top-level writes, preserving each JSON value's type.
    /// Keys in either collection must not contain U+0000.
    pub set: BTreeMap<String, Value>,
    /// Install or extend this expiry; `None` preserves the stored expiry.
    pub extend_expires_at: Option<i64>,
    /// Applied as the maximum of the stored and supplied timestamps.
    pub updated_at: i64,
}

/// Keyset pagination boundary over the notes store's default total order
/// (`created_at DESC, id ASC` — see `note_filter_page_order_clause`).
///
/// Only meaningful when [`NoteFilter::order_by`] is `None`: the boundary is
/// expressed in terms of that default order, and combining it with a custom
/// sort field has no defined meaning, so callers must not set both.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct NoteSeekAfter {
    pub created_at: i64,
    pub id: Uuid,
}

/// Boundary in an ascending RFC 3339 property order. Equal instants sort by
/// their stored text and then note ID, so this cursor identifies one row even
/// when distinct offsets encode the same instant.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct NoteInstantSeekAfter {
    pub value: String,
    pub id: Uuid,
}

#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum NoteTagMode {
    #[default]
    Any,
    All,
}

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct NoteKeyCursor {
    pub updated_at: i64,
    pub key: String,
    pub id: Uuid,
}

impl From<&Note> for NoteKeyCursor {
    fn from(note: &Note) -> Self {
        Self {
            updated_at: note.updated_at,
            key: note.key.clone().expect("keyed row"),
            id: note.id,
        }
    }
}

/// Inclusive expiry cutoff with a creation cutoff only for notes without an expiry.
/// Both bounds use microseconds since the Unix epoch.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct NoteExpiryFallback {
    pub expires_at_or_before: i64,
    pub created_at_or_before: i64,
}

/// Ascending native timestamp order, with ascending note ID as the final tie-break.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NoteTimeOrder {
    /// `expires_at ASC, id ASC`; absent expiries sort first.
    ExpiresAt,
    /// `COALESCE(expires_at, created_at) ASC, id ASC`.
    ExpiresAtOrCreatedAt,
}

/// Filter + sort options for [`NoteStore::query_notes_filtered`].
///
/// Designed for general property-based filtering on any JSON field, not
/// schedule-specific, so D9 and future packs can reuse the same API.
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct NoteFilter {
    pub kind: Option<String>,
    #[serde(default)]
    pub property_filters: Vec<PropertyFilter>,
    /// `(json_path, direction)` — `None` defaults to `created_at DESC`.
    pub order_by: Option<(String, SortDir)>,
    /// Interpret `order_by`'s text property as an RFC 3339 UTC instant before
    /// sorting. Only ascending order is currently supported.
    #[serde(default)]
    pub order_by_instant: bool,
    /// When true, omit the SQL ordering clause from count-free pages. The
    /// caller is responsible for ordering the bounded result set.
    #[serde(default)]
    pub unordered: bool,
    /// When non-empty, restricts results to any of these namespaces using
    /// `namespace IN (...)`. Takes precedence over the `namespace` string
    /// parameter passed to `query_notes_filtered`. When empty the
    /// caller-supplied `namespace` parameter is used (backward-compatible).
    #[serde(default)]
    pub namespaces: Vec<String>,
    /// Restrict to notes where `created_at >= min_created_at` (microseconds epoch).
    /// `None` applies no lower-bound constraint.
    pub min_created_at: Option<i64>,
    /// Inclusive creation-time upper bound in microseconds since the Unix epoch.
    #[serde(default)]
    pub max_created_at: Option<i64>,
    /// Inclusive expiry-time upper bound in microseconds since the Unix epoch.
    /// Rows with no expiry do not match a supplied bound.
    #[serde(default)]
    pub max_expires_at: Option<i64>,
    /// Include soft-deleted rows in filtered reads. Property patches remain live-only.
    #[serde(default)]
    pub include_deleted: bool,
    /// Select expired notes, falling back to creation time only when expiry is absent.
    /// This predicate is ANDed with all other bounds and property filters.
    #[serde(default)]
    pub expiry_fallback: Option<NoteExpiryFallback>,
    /// Native ascending timestamp order for counted/count-free/bounded pages.
    /// Incompatible with property/instant/unordered ordering or any cursor.
    /// Keyed/sequence reads and filtered mutations reject it; counts ignore ordering.
    #[serde(default)]
    pub time_order: Option<NoteTimeOrder>,
    #[serde(default)]
    pub min_updated_at: Option<i64>,
    #[serde(default)]
    pub tags: Vec<String>,
    #[serde(default)]
    pub tag_mode: NoteTagMode,
    /// Restrict results to rows strictly after this boundary in the default
    /// `created_at DESC, id ASC` order, for keyset (seek) pagination that
    /// avoids re-walking earlier pages the way `PageRequest.offset` does.
    /// Requires `order_by` to be `None`.
    ///
    /// Honoured only by [`NoteStore::query_notes_filtered_count_free`], which
    /// seeks directly to the boundary and returns `total: None`.
    /// [`NoteStore::query_notes_filtered`] rejects a non-`None` value with
    /// `StorageError::InvalidInput`: it computes an exact `COUNT(*)` total
    /// over the whole matching set, which has no defined meaning paired with
    /// a seek boundary.
    #[serde(default)]
    pub after: Option<NoteSeekAfter>,
    /// Seek after a row in the `order_by_instant` total order. Honoured only
    /// by `query_notes_filtered_count_free` with `offset: 0`.
    #[serde(default)]
    pub after_instant: Option<NoteInstantSeekAfter>,
    /// Restrict `message` rows to the ones this mailbox reader may see, in
    /// the query itself. Rows of every other kind are unaffected. Applying the
    /// partition here means a scan window, and any cursor derived from it,
    /// only ever contains rows the reader is allowed to see.
    #[serde(default)]
    pub mailbox: Option<NoteMailboxScope>,
}

impl NoteFilter {
    /// Select expiry timestamps at or before `micros` (inclusive, Unix microseconds).
    /// A note without an expiry does not match this bound.
    pub fn expires_before(mut self, micros: i64) -> Self {
        self.max_expires_at = Some(micros);
        self
    }

    /// Select creation timestamps at or before `micros` (inclusive, Unix microseconds).
    pub fn created_before(mut self, micros: i64) -> Self {
        self.max_created_at = Some(micros);
        self
    }

    /// Include tombstones in filtered reads without permitting property patches on them.
    pub fn include_deleted(mut self) -> Self {
        self.include_deleted = true;
        self
    }
}

/// The actor partition a mailbox reader may see, as evaluated by the store.
/// It mirrors the row-level mailbox rule: an inbound message belongs to its
/// `to_actor`, an outbound message to its `from_actor`, and `legacy_local`
/// additionally admits the unattributed pre-routing rows.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct NoteMailboxScope {
    pub actor_id: String,
    pub legacy_local: bool,
}

/// Temporal-referential note CRUD over the notes substrate table.
#[async_trait]
pub trait NoteStore: Send + Sync + 'static {
    async fn get_live_notes_by_key(
        &self,
        _namespace: &str,
        _key: &str,
        _kind: Option<&str>,
    ) -> StorageResult<Vec<Note>> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "get_live_notes_by_key".into(),
            message: "backend has no keyed note lookup".into(),
        })
    }

    async fn query_keyed_notes(
        &self,
        _namespace: &str,
        _filter: &NoteFilter,
        _prefix: &str,
        _after: Option<&NoteKeyCursor>,
        _page: PageRequest,
    ) -> StorageResult<(Vec<Note>, Option<NoteKeyCursor>)> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "query_keyed_notes".into(),
            message: "backend has no keyed note pagination".into(),
        })
    }

    /// Insert or update a single note. Updates preserve the stored immutable key.
    async fn upsert_note(&self, note: Note) -> StorageResult<()>;
    /// Replace a note only when the persisted row still matches the caller's
    /// read snapshot.
    ///
    /// `expected_updated_at` is the snapshot revision and
    /// `expected_deleted_at` closes the soft-delete race (legacy soft-delete
    /// paths may change `deleted_at` without changing `updated_at`). The
    /// replacement note's `updated_at` must be strictly greater than that
    /// persisted revision. Returns `false` when the row disappeared, changed,
    /// or was supplied a non-advancing replacement revision. This is the
    /// full-note compare-and-swap seam used when a pack hook derives coupled
    /// fields from that snapshot before persistence. The default returns
    /// `Unsupported` rather than falling back to an unguarded upsert and
    /// reintroducing the stale-snapshot race.
    async fn replace_note_if_unchanged(
        &self,
        _note: Note,
        _expected_updated_at: i64,
        _expected_deleted_at: Option<i64>,
    ) -> StorageResult<bool> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "replace_note_if_unchanged".into(),
            message: "this backend does not implement guarded note replacement".into(),
        })
    }
    /// Insert a note only if no row already holds its id, reporting whether
    /// this call is the one that inserted it.
    ///
    /// This closes the other half of the read-modify-write race that
    /// [`NoteStore::replace_note_if_unchanged`] closes. That one protects a
    /// caller who read an existing row; this one protects a caller who read
    /// *no* row. Where the id is derived rather than freshly generated — a
    /// deterministic per-subject id — two callers can both read absence and
    /// both write, and an upsert resolves that by overwriting, so the first
    /// caller's write is lost with no error on either side. Returns `false`
    /// when a row already existed, which the caller maps to a conflict rather
    /// than to success.
    ///
    /// The pre-existing row is left exactly as it is: this must not be
    /// implemented as an upsert, since overwriting is the behaviour being
    /// avoided. The default returns `Unsupported` for that reason, rather
    /// than falling back to `upsert_note` and silently reintroducing the
    /// race under a name that promises otherwise.
    async fn insert_note_if_absent(&self, _note: Note) -> StorageResult<bool> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "insert_note_if_absent".into(),
            message: "this backend does not implement guarded note insertion".into(),
        })
    }
    /// Insert or update a batch of notes. Updates preserve each stored immutable key.
    async fn upsert_notes(&self, notes: Vec<Note>) -> StorageResult<BatchWriteSummary>;
    /// Fetch a note by UUID, returning `None` if absent.
    async fn get_note(&self, id: Uuid) -> StorageResult<Option<Note>>;
    /// Fetch a note by UUID regardless of soft-deletion state.
    ///
    /// Returns the note row even when `deleted_at` is set. Callers use this
    /// to distinguish "soft-deleted" from "never existed".
    async fn get_note_including_deleted(&self, id: Uuid) -> StorageResult<Option<Note>>;
    /// Delete a note by UUID using the specified delete mode.
    async fn delete_note(&self, id: Uuid, mode: DeleteMode) -> StorageResult<bool>;
    /// Patch `properties`/`updated_at` on an existing note in place via a real
    /// `UPDATE`, leaving every other column (including the row's `rowid`)
    /// untouched.
    ///
    /// Unlike `upsert_note`, which writes the complete note shape, this leaves
    /// every non-property column untouched. It also never churns the row's
    /// implicit `rowid`, which is required by callers relying on stable row
    /// identity (#780).
    /// Returns `true` when a live (non-soft-deleted) row with this `id` was
    /// found and updated, `false` otherwise.
    async fn update_note_properties(
        &self,
        id: Uuid,
        properties: Option<Value>,
        updated_at: i64,
    ) -> StorageResult<bool>;
    /// Atomically set one top-level key in a note's JSON `properties` object.
    ///
    /// The backend must perform the read/modify/write as one storage operation
    /// so concurrent writes to different keys cannot overwrite each other.
    /// `value` keeps its JSON type, including explicit JSON `null`. A SQL-NULL
    /// property document is initialized as an empty object. A live row whose
    /// stored document is a non-object is not modified and returns `false`, as
    /// do missing and soft-deleted rows. Keys containing U+0000 must be
    /// rejected: SQLite JSON-path labels cannot address them without risking
    /// mutation of a shorter sibling key.
    async fn set_note_property(
        &self,
        id: Uuid,
        key: &str,
        value: Value,
        updated_at: i64,
    ) -> StorageResult<bool>;
    /// Atomically patch a single `properties` JSON key on a note, but only
    /// when the row's *current* state (re-evaluated inside this same
    /// statement, not a snapshot the caller fetched earlier) still satisfies
    /// `filter`'s namespace/kind/property_filters.
    ///
    /// Unlike `set_note_property` (which patches unconditionally once the row
    /// is live) or `update_note_properties` (which replaces the whole
    /// `properties` column with a value the caller already computed — safe
    /// only when nothing else can have written to the row since the caller's
    /// read), this also rechecks `filter` against the row's live state before
    /// writing, so a target that stopped matching an eligibility predicate
    /// between validation and this call is not mutated. Any other property
    /// written concurrently between the caller's read and this call survives
    /// untouched either way. A live row whose stored `properties` document is
    /// a non-object (scalar, array, or otherwise) is not modified and returns
    /// `false`, mirroring `set_note_property`. Returns `Ok(false)` — not an
    /// error — when no live row currently matches `filter` (id not found,
    /// soft-deleted, an eligibility property changed since the caller last
    /// validated it, or the stored document is not a JSON object); the
    /// caller degrades that the same way as `update_note_properties`'s
    /// `Ok(false)`.
    async fn try_patch_note_property(
        &self,
        id: Uuid,
        namespace: &str,
        filter: &NoteFilter,
        json_path: &str,
        value: Value,
        updated_at: i64,
    ) -> StorageResult<bool>;
    /// Atomically apply a bounded partial property update to one live note.
    ///
    /// The current row must match the id, exact namespace, exact kind, and all
    /// preconditions in the same operation that merges the writes. Namespace
    /// is an equality guard, not an authorization boundary. A SQL-NULL property
    /// document starts as `{}`; non-object documents and any other nonmatch
    /// return `false` without changing the row. Unrelated properties, immutable
    /// key, row identity, and non-target columns survive. A matched update
    /// advances the persisted version, even when the supplied timestamp is older.
    ///
    /// Validate bounds, keys and write values before writer admission. Maintain
    /// derived retry-deadline fields when `next_attempt_at` is written, and leave
    /// them untouched otherwise. Unsupported backends must fail closed without
    /// a read/replace or upsert fallback.
    async fn try_patch_note_properties(
        &self,
        _id: Uuid,
        _namespace: &str,
        _kind: &str,
        _patch: &NotePropertyPatch,
    ) -> StorageResult<bool> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "try_patch_note_properties".into(),
            message: "this backend does not implement guarded partial note updates".into(),
        })
    }
    /// Atomically patch one JSON property on every supplied note.
    ///
    /// Each target is rechecked against `namespace` and `filter` inside the
    /// same transaction. The operation commits only when every distinct id
    /// matches exactly one live object-valued row; a missing, soft-deleted, or
    /// no-longer-eligible target rolls the entire unit back. Other property
    /// keys are preserved by the same storage-side `json_set` operation as
    /// [`Self::try_patch_note_property`]. Backends without a transactional
    /// multi-note implementation retain the default `Unsupported` result.
    async fn patch_note_property_atomic(
        &self,
        _ids: Vec<Uuid>,
        _namespace: &str,
        _filter: &NoteFilter,
        _json_path: &str,
        _value: Value,
        _updated_at: i64,
    ) -> StorageResult<()> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "patch_note_property_atomic".into(),
            message: "this backend does not implement atomic multi-note property patches".into(),
        })
    }
    /// Query notes by namespace and optional kind with pagination.
    /// The returned total and page items must come from one consistent
    /// backend snapshot.
    async fn query_notes(
        &self,
        namespace: &str,
        kind: Option<&str>,
        page: PageRequest,
    ) -> StorageResult<Page<Note>>;
    /// Query notes by namespace and optional kind without computing an exact total.
    ///
    /// Implementations must return `total: None`. Backends that cannot
    /// guarantee a count-free projection must fail closed rather than invoke
    /// [`Self::query_notes`], whose exact total remains available to callers
    /// that expose a snapshot-consistent count.
    async fn query_notes_count_free(
        &self,
        _namespace: &str,
        _kind: Option<&str>,
        _page: PageRequest,
    ) -> StorageResult<Page<Note>> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "query_notes_count_free".into(),
            message: "this backend does not implement count-free note paging".into(),
        })
    }
    /// Query notes with property-based filtering and custom sort.
    /// The returned total and page items must come from one consistent
    /// backend snapshot.
    async fn query_notes_filtered(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        page: PageRequest,
    ) -> StorageResult<Page<Note>>;
    /// Query a filtered note page without computing an exact total.
    ///
    /// Implementations must return `total: None`; callers that need a
    /// `has_more` bit should request one lookahead row. Backends that cannot
    /// guarantee a count-free projection must fail closed rather than invoke
    /// [`Self::query_notes_filtered`], whose exact total is intentionally
    /// backlog-proportional.
    async fn query_notes_filtered_count_free(
        &self,
        _namespace: &str,
        _filter: &NoteFilter,
        _page: PageRequest,
    ) -> StorageResult<Page<Note>> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "query_notes_filtered_count_free".into(),
            message: "this backend does not implement count-free filtered note paging".into(),
        })
    }
    /// Count several filtered note populations in one consistent backend
    /// snapshot. Backends that cannot provide that guarantee must return
    /// [`crate::StorageError::Unsupported`] rather than composing independent
    /// queries or treating an absent page total as zero. SQL backends should
    /// override this operation so callers can retain separate index-friendly
    /// predicates without racing between their counts.
    async fn count_notes_filtered_in_snapshot(
        &self,
        _namespace: &str,
        _filters: &[NoteFilter],
    ) -> StorageResult<Vec<u64>> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "count_notes_filtered_in_snapshot".into(),
            message: "this backend does not implement snapshot-consistent filtered note counts"
                .into(),
        })
    }
    /// Count several filtered note populations in one backend snapshot, with
    /// each count bounded by `cap`.
    ///
    /// A saturated entry reports `count == cap` and `saturated == true`; an
    /// unsaturated entry is exact. SQL implementations should count over a
    /// limited `SELECT 1` subquery so work is proportional to `cap` and the
    /// caller's own matching population, never an unrelated backlog. That
    /// bound holds only when every `property_filter` in the given
    /// `NoteFilter` is servable as an index key column (equality on a bound
    /// parameter or a plan-time-provable partial-index predicate) all the way
    /// down to the filter that actually narrows the population — a residual
    /// filter evaluated after the index seek (a bound parameter the index
    /// cannot use to skip rows) degrades the bound to work proportional to
    /// the rows *scanned* before `cap` matches are found, not the rows that
    /// match. Callers building filters against this bound must ensure the
    /// full predicate set is indexed, not just the predicate that narrows the
    /// population most. Backends that cannot preserve the shared snapshot or
    /// work bound must fail closed.
    async fn count_notes_filtered_bounded_in_snapshot(
        &self,
        _namespace: &str,
        _filters: &[NoteFilter],
        _cap: u32,
    ) -> StorageResult<Vec<BoundedCount>> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "count_notes_filtered_bounded_in_snapshot".into(),
            message: "this backend does not implement snapshot-consistent bounded note counts"
                .into(),
        })
    }
    /// Resolve a note id to its immutable insertion sequence.
    async fn note_sequence(&self, _id: Uuid) -> StorageResult<Option<i64>> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "note_sequence".into(),
            message: "this backend does not implement note insertion sequences".into(),
        })
    }
    /// Query an immutable insertion-sequence keyset page with the same
    /// predicates as [`Self::query_notes_filtered`].
    async fn query_notes_filtered_after(
        &self,
        _namespace: &str,
        _filter: &NoteFilter,
        _after: Option<SeekCursor>,
        _limit: u32,
    ) -> StorageResult<SeekPage<Note>> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "query_notes_filtered_after".into(),
            message: "this backend does not implement note seek pagination".into(),
        })
    }
    /// Fetch up to `max_rows + 1` notes matching `filter` in a single
    /// deterministically-ordered SQL statement, with no separate `COUNT(*)`
    /// and no pagination loop.
    ///
    /// A single statement observes one consistent snapshot for its entire
    /// execution, so the result cannot be split across a concurrent insert
    /// the way a `COUNT(*)` followed by independent `LIMIT`/`OFFSET` pages
    /// can. Callers detect the over-bound case by checking whether the
    /// returned `Vec` has more than `max_rows` items — that means at least
    /// `max_rows + 1` rows matched and the caller must reject the query
    /// rather than silently return a truncated, possibly priority-incomplete
    /// set.
    async fn query_notes_filtered_bounded(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        max_rows: u32,
    ) -> StorageResult<Vec<Note>>;
    /// Count notes in a namespace, optionally filtered by kind.
    async fn count_notes(&self, namespace: &str, kind: Option<&str>) -> StorageResult<u64>;
    /// Count notes across the given namespaces, optionally filtered by kind.
    /// The default preserves compatibility by summing the existing
    /// single-namespace operation; SQL backends should override this with one
    /// `IN` aggregate.
    async fn count_notes_in_namespaces(
        &self,
        namespaces: &[String],
        kind: Option<&str>,
    ) -> StorageResult<u64> {
        let mut total = 0;
        for namespace in namespaces {
            total += self.count_notes(namespace, kind).await?;
        }
        Ok(total)
    }

    /// Attempt to insert a note without overwriting an existing row.
    ///
    /// Returns `true` when the row was newly written.  Returns `false` only
    /// when a live note with the same non-empty `external_id` already exists in
    /// the same namespace and kind (confirmed dedup hit).  Any other constraint
    /// violation (e.g. a primary key collision) is surfaced as a `StorageError`
    /// so that callers do not misinterpret unexpected failures as deduplication.
    async fn try_insert_note(&self, note: Note) -> StorageResult<bool>;

    /// Atomically insert an ingest note and the blob attachments owned by it.
    /// A deduplicated note writes no attachments; callers must inspect the
    /// existing note before repairing a legacy missing role.
    async fn try_insert_note_with_attachments(
        &self,
        _note: Note,
        _attachments: Vec<crate::Attachment>,
    ) -> StorageResult<bool> {
        Err(crate::StorageError::Unsupported {
            capability: crate::StorageCapability::Notes,
            operation: "try_insert_note_with_attachments".into(),
            message: "this backend cannot atomically insert note attachments".into(),
        })
    }

    /// Fetch multiple notes by UUID in a single call.
    async fn get_notes_batch(&self, ids: &[Uuid]) -> StorageResult<Vec<Note>> {
        let mut out = Vec::with_capacity(ids.len());
        for &id in ids {
            if let Some(n) = self.get_note(id).await? {
                out.push(n);
            }
        }
        Ok(out)
    }

    /// Fetch multiple notes, including soft-deleted rows, for mutation policy checks.
    /// Missing IDs are omitted; callers correlate rows by ID rather than result order.
    async fn get_notes_batch_including_deleted(&self, ids: &[Uuid]) -> StorageResult<Vec<Note>> {
        let mut out = Vec::with_capacity(ids.len());
        for &id in ids {
            if let Some(note) = self.get_note_including_deleted(id).await? {
                out.push(note);
            }
        }
        Ok(out)
    }

    /// Fetch only the columns needed for note visibility checks by UUID.
    /// Missing IDs are omitted; soft-deleted rows retain their deletion time.
    async fn get_note_visibility_batch(&self, ids: &[Uuid]) -> StorageResult<Vec<NoteVisibility>> {
        let mut out = Vec::with_capacity(ids.len());
        for &id in ids {
            if let Some(note) = self.get_note_including_deleted(id).await? {
                out.push(NoteVisibility {
                    id: note.id,
                    namespace: note.namespace,
                    deleted_at: note.deleted_at,
                });
            }
        }
        Ok(out)
    }
}