khive-runtime 0.10.0

//! Policy-enforcing decorator around the [`NoteStore`] returned by the public
//! [`crate::KhiveRuntime::notes`] accessor.
//!
//! `comm.health` (`khive-pack-comm`) counts every live `kind = "message"` row
//! whose JSON `properties` carries a truthy `quarantined` value, grouped by
//! `channel_kind`/`channel_slug`. Those three keys are transport-owned
//! evidence that only the trusted channel-ingest path
//! (`KhiveRuntime::try_create_note_as_trusted_ingest`) may establish. The
//! generic `create`/`update` verb funnel and the proposal-apply path already
//! run a pack-installed note-write validator that refuses them, but that
//! validator does not cover a caller reaching `notes(&token)` directly and
//! calling the raw storage insert/upsert methods — this decorator closes
//! that seam by refusing the same three properties on a `kind = "message"`
//! note at every full-note write. Existing non-message kinds with kind-owned
//! identity (currently `channel_health`) may retain those fields on a full-row
//! write, but cannot change or remove them. This allows heartbeat metadata/CAS
//! updates through the public accessor while preserving its channel coordinates.
//!
//! The boundary this decorator enforces is the runtime's TYPED accessor
//! surface — the seams pack code actually reaches. `KhiveRuntime::backend()`
//! and `KhiveRuntime::sql()` are embedder capabilities (pools, schema plans,
//! diagnostics, raw SQL): an embedder holding them already holds
//! root-equivalent access to the database file itself, so no store-layer
//! check can bind it. No pack code obtains `NoteStore` objects from
//! `backend()` (the integration suite pins where the boundary sits). Two
//! packs do issue note DML through `sql()` as deliberate low-level writers:
//! gtd's atomic task transition (`gtd_transition_statement` — the handler
//! loads the row and verifies `kind == "task"` first, and the UPDATE pins
//! the row to that verified snapshot via its `updated_at` optimistic
//! guard) and schedule's event updates (handler pre-check plus
//! `kind = 'scheduled_event'` in the SQL itself). Both are therefore held
//! off `kind = "message"` transport evidence; widening either constraint,
//! or adding a third such writer, must instead route through the typed
//! accessor this module wraps.
//!
//! `try_create_note_impl` (the runtime-internal implementation backing both
//! `try_create_note` and `try_create_note_as_trusted_ingest`) bypasses this
//! decorator entirely by reaching the backend directly
//! (`KhiveRuntime::raw_notes`) rather than through `notes()`; it enforces the
//! identical property check itself (over the same kind-owned list this
//! decorator reads), conditionally allowing those keys only when called
//! with a `ChannelIngestCapability`. That keeps the trusted
//! ingest path exempt by construction rather than by call site.

use std::collections::HashMap;
use std::sync::Arc;

use async_trait::async_trait;
use khive_storage::{
    BatchWriteSummary, BoundedCount, DeleteMode, Note, NoteFilter, NoteStore, NoteVisibility, Page,
    PageRequest, SeekCursor, SeekPage, StorageCapability, StorageError, StorageResult,
};
use serde_json::Value;
use uuid::Uuid;

use crate::curation::kind_owned_properties;

fn transport_owned_message_property_named_in(
    properties: &serde_json::Map<String, Value>,
) -> Option<&'static str> {
    kind_owned_properties("message")
        .iter()
        .copied()
        .find(|key| properties.contains_key(*key))
}

fn reject_if_forged_message_note(note: &Note, operation: &'static str) -> StorageResult<()> {
    if note.kind != "message" {
        return Ok(());
    }
    let Some(key) = note
        .properties
        .as_ref()
        .and_then(Value::as_object)
        .and_then(transport_owned_message_property_named_in)
    else {
        return Ok(());
    };
    Err(StorageError::InvalidInput {
        capability: StorageCapability::Notes,
        operation: operation.into(),
        message: format!(
            "`{key}` is transport-owned on a `message` note and cannot be written through the \
             public NoteStore accessor; only the trusted channel-ingest path may establish \
             quarantine disposition and channel provenance"
        ),
    })
}

fn has_web_receipt_provenance(note: &Note) -> bool {
    note.properties.as_ref().is_some_and(|properties| {
        properties
            .as_object()
            .is_some_and(|map| map.contains_key(crate::secret_gate::RESERVED_WEB_RECEIPT_KEY))
    })
}

fn web_receipt_write_refused(operation: &'static str) -> StorageError {
    StorageError::InvalidInput {
        capability: StorageCapability::Notes,
        operation: operation.into(),
        message: "web receipt provenance and its receipt body are web-pack-owned; generic note writes cannot create or alter them".into(),
    }
}

fn reject_forged_web_receipt_note(note: &Note, operation: &'static str) -> StorageResult<()> {
    if has_web_receipt_provenance(note) {
        return Err(web_receipt_write_refused(operation));
    }
    Ok(())
}

/// The property-patch seams cannot see the note's kind in their signatures,
/// so they refuse the reserved keys unconditionally: no public-store caller
/// legitimately patches transport-owned keys on any note kind (quarantine
/// disposition is established only at ingest, through the capability path),
/// and a kind-scoped check would need a lookup whose result the refusal
/// would then depend on. A future release-from-quarantine flow belongs on
/// the capability path, not here.
///
/// The guard accepts only targets whose top-level segment it can fully
/// parse: an optional `$` root, an optional `.` separator, then a bare
/// identifier. SQLite's JSON path grammar admits other spellings of the
/// same top-level key — quoted labels (`$."quarantined"`), bracket and
/// escaped forms, and the bare `$` root that replaces the whole properties
/// object — which a substring split cannot canonicalize, so every such
/// spelling is refused outright rather than compared against the reserved
/// list: a target this guard cannot prove innocent does not pass it. The
/// same rule covers `set_note_property` keys, which the store interpolates
/// into a quoted JSON path. Production callers patch only bare-identifier
/// targets (`read`, `$.read`, `external_id`), so the strictness costs no
/// legitimate caller anything.
fn reject_reserved_patch_target(target: &str, operation: &'static str) -> StorageResult<()> {
    let body = target.strip_prefix('$').unwrap_or(target);
    let body = body.strip_prefix('.').unwrap_or(body);
    let first_segment = &body[..body.find(['.', '[']).unwrap_or(body.len())];
    let is_bare_identifier = !first_segment.is_empty()
        && !first_segment.starts_with(|c: char| c.is_ascii_digit())
        && first_segment
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '_');
    if !is_bare_identifier {
        return Err(StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: operation.into(),
            message: format!(
                "property target `{target}` is not a bare top-level identifier; the public \
                 NoteStore accessor refuses target spellings it cannot prove distinct from \
                 the transport-owned message properties"
            ),
        });
    }
    if !kind_owned_properties("message").contains(&first_segment) {
        return Ok(());
    }
    Err(StorageError::InvalidInput {
        capability: StorageCapability::Notes,
        operation: operation.into(),
        message: format!(
            "`{first_segment}` is transport-owned and cannot be patched through the public \
             NoteStore accessor; only the trusted channel-ingest path may establish quarantine \
             disposition and channel provenance"
        ),
    })
}

fn reject_reserved_replacement_properties(
    properties: Option<&Value>,
    operation: &'static str,
) -> StorageResult<()> {
    if properties
        .and_then(Value::as_object)
        .is_some_and(|map| map.contains_key(crate::secret_gate::RESERVED_WEB_RECEIPT_KEY))
    {
        return Err(web_receipt_write_refused(operation));
    }
    let Some(key) = properties
        .and_then(Value::as_object)
        .and_then(transport_owned_message_property_named_in)
    else {
        return Ok(());
    };
    Err(StorageError::InvalidInput {
        capability: StorageCapability::Notes,
        operation: operation.into(),
        message: format!(
            "`{key}` is transport-owned and cannot be written through the public NoteStore \
             accessor; only the trusted channel-ingest path may establish quarantine disposition \
             and channel provenance"
        ),
    })
}

fn reject_reserved_note_properties(
    properties: Option<&Value>,
    operation: &'static str,
) -> StorageResult<()> {
    crate::secret_gate::reject_reserved_secret_gate_property(properties).map_err(|error| {
        StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: operation.into(),
            message: error.to_string(),
        }
    })
}

fn reject_existing_secret_gate_property(
    existing: Option<&Note>,
    operation: &'static str,
) -> StorageResult<()> {
    if let Some(existing) = existing {
        reject_reserved_note_properties(existing.properties.as_ref(), operation)?;
    }
    Ok(())
}

fn reject_changed_identity_properties(
    existing: &Note,
    properties: Option<&Value>,
    operation: &'static str,
) -> StorageResult<()> {
    // Message admission remains presence-based, with its existing error text.
    if existing.kind == "message" {
        return Ok(());
    }
    for key in kind_owned_properties(&existing.kind) {
        let before = existing
            .properties
            .as_ref()
            .and_then(|value| value.get(*key));
        let after = properties.and_then(|value| value.get(*key));
        if before != after {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: operation.into(),
                message: format!(
                    "`{key}` is kind-owned on a `{}` note and cannot be changed through the \
                     public NoteStore accessor",
                    existing.kind
                ),
            });
        }
    }
    Ok(())
}

fn reject_changed_identity_note(
    existing: &Note,
    note: &Note,
    operation: &'static str,
) -> StorageResult<()> {
    if has_web_receipt_provenance(existing) {
        return Err(web_receipt_write_refused(operation));
    }
    if existing.kind != "message"
        && !kind_owned_properties(&existing.kind).is_empty()
        && existing.kind != note.kind
    {
        // Otherwise a caller could remove the owning kind first and change its
        // coordinates in the next public write.
        return Err(StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: operation.into(),
            message: format!(
                "the kind of a `{}` note with kind-owned identity cannot be changed through \
                 the public NoteStore accessor",
                existing.kind
            ),
        });
    }
    reject_changed_identity_properties(existing, note.properties.as_ref(), operation)
}

/// Wraps `inner` so every full-note insert/upsert seam AND every
/// property-patch seam enforces the reserved-transport-property policy
/// described at module level. Message writes retain their presence-based
/// refusal. Other kind-owned identities are compared with the existing row,
/// including tombstones, before replacement. Patch targets retain their existing
/// transport-key refusal; whole-property replacement also cannot erase identity.
pub(crate) struct PolicyEnforcingNoteStore {
    inner: Arc<dyn NoteStore>,
}

impl PolicyEnforcingNoteStore {
    pub(crate) fn wrap(inner: Arc<dyn NoteStore>) -> Arc<dyn NoteStore> {
        Arc::new(Self { inner })
    }

    async fn reject_identity_change(
        &self,
        note: &Note,
        operation: &'static str,
    ) -> StorageResult<()> {
        if let Some(existing) = self.inner.get_note_including_deleted(note.id).await? {
            reject_reserved_note_properties(existing.properties.as_ref(), operation)?;
            reject_changed_identity_note(&existing, note, operation)?;
        }
        Ok(())
    }

    async fn reject_web_receipt_mutation(
        &self,
        id: Uuid,
        operation: &'static str,
    ) -> StorageResult<()> {
        if self
            .inner
            .get_note_including_deleted(id)
            .await?
            .as_ref()
            .is_some_and(has_web_receipt_provenance)
        {
            return Err(web_receipt_write_refused(operation));
        }
        Ok(())
    }
}

#[async_trait]
impl NoteStore for PolicyEnforcingNoteStore {
    async fn get_live_notes_by_key(
        &self,
        namespace: &str,
        key: &str,
        kind: Option<&str>,
    ) -> StorageResult<Vec<Note>> {
        self.inner.get_live_notes_by_key(namespace, key, kind).await
    }

    async fn query_keyed_notes(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        prefix: &str,
        after: Option<&khive_storage::note::NoteKeyCursor>,
        page: PageRequest,
    ) -> StorageResult<(Vec<Note>, Option<khive_storage::note::NoteKeyCursor>)> {
        self.inner
            .query_keyed_notes(namespace, filter, prefix, after, page)
            .await
    }

    async fn upsert_note(&self, note: Note) -> StorageResult<()> {
        reject_reserved_note_properties(note.properties.as_ref(), "upsert_note")?;
        reject_if_forged_message_note(&note, "upsert_note")?;
        reject_forged_web_receipt_note(&note, "upsert_note")?;
        self.reject_identity_change(&note, "upsert_note").await?;
        self.inner.upsert_note(note).await
    }

    async fn insert_note_if_absent(&self, note: Note) -> StorageResult<bool> {
        reject_reserved_note_properties(note.properties.as_ref(), "insert_note_if_absent")?;
        reject_if_forged_message_note(&note, "insert_note_if_absent")?;
        reject_forged_web_receipt_note(&note, "insert_note_if_absent")?;
        self.inner.insert_note_if_absent(note).await
    }

    async fn replace_note_if_unchanged(
        &self,
        note: Note,
        expected_updated_at: i64,
        expected_deleted_at: Option<i64>,
    ) -> StorageResult<bool> {
        reject_reserved_note_properties(note.properties.as_ref(), "replace_note_if_unchanged")?;
        reject_if_forged_message_note(&note, "replace_note_if_unchanged")?;
        if !has_web_receipt_provenance(&note) {
            self.reject_web_receipt_mutation(note.id, "replace_note_if_unchanged")
                .await?;
        }
        reject_forged_web_receipt_note(&note, "replace_note_if_unchanged")?;
        self.reject_identity_change(&note, "replace_note_if_unchanged")
            .await?;
        self.inner
            .replace_note_if_unchanged(note, expected_updated_at, expected_deleted_at)
            .await
    }

    async fn upsert_notes(&self, notes: Vec<Note>) -> StorageResult<BatchWriteSummary> {
        for note in &notes {
            reject_reserved_note_properties(note.properties.as_ref(), "upsert_notes")?;
            reject_if_forged_message_note(note, "upsert_notes")?;
            reject_forged_web_receipt_note(note, "upsert_notes")?;
        }
        {
            // Validate every row before the batch write. Repeated IDs must also
            // preserve identity established earlier in this same batch.
            let mut preceding: HashMap<Uuid, &Note> = HashMap::new();
            for note in &notes {
                if let Some(existing) = preceding.get(&note.id) {
                    reject_changed_identity_note(existing, note, "upsert_notes")?;
                } else {
                    self.reject_identity_change(note, "upsert_notes").await?;
                }
                preceding.insert(note.id, note);
            }
        }
        self.inner.upsert_notes(notes).await
    }

    async fn get_note(&self, id: Uuid) -> StorageResult<Option<Note>> {
        self.inner.get_note(id).await
    }

    async fn get_note_including_deleted(&self, id: Uuid) -> StorageResult<Option<Note>> {
        self.inner.get_note_including_deleted(id).await
    }

    async fn delete_note(&self, id: Uuid, mode: DeleteMode) -> StorageResult<bool> {
        self.inner.delete_note(id, mode).await
    }

    async fn update_note_properties(
        &self,
        id: Uuid,
        properties: Option<Value>,
        updated_at: i64,
    ) -> StorageResult<bool> {
        reject_reserved_note_properties(properties.as_ref(), "update_note_properties")?;
        reject_reserved_replacement_properties(properties.as_ref(), "update_note_properties")?;
        self.reject_web_receipt_mutation(id, "update_note_properties")
            .await?;
        if let Some(existing) = self.inner.get_note_including_deleted(id).await? {
            reject_reserved_note_properties(
                existing.properties.as_ref(),
                "update_note_properties",
            )?;
            reject_changed_identity_properties(
                &existing,
                properties.as_ref(),
                "update_note_properties",
            )?;
        }
        self.inner
            .update_note_properties(id, properties, updated_at)
            .await
    }

    async fn set_note_property(
        &self,
        id: Uuid,
        key: &str,
        value: Value,
        updated_at: i64,
    ) -> StorageResult<bool> {
        reject_reserved_patch_target(key, "set_note_property")?;
        let existing = self.inner.get_note_including_deleted(id).await?;
        if existing.as_ref().is_some_and(has_web_receipt_provenance) {
            return Err(web_receipt_write_refused("set_note_property"));
        }
        reject_existing_secret_gate_property(existing.as_ref(), "set_note_property")?;
        self.inner
            .set_note_property(id, key, value, updated_at)
            .await
    }

    async fn try_patch_note_property(
        &self,
        id: Uuid,
        namespace: &str,
        filter: &NoteFilter,
        json_path: &str,
        value: Value,
        updated_at: i64,
    ) -> StorageResult<bool> {
        reject_reserved_patch_target(json_path, "try_patch_note_property")?;
        let existing = self.inner.get_note_including_deleted(id).await?;
        if existing.as_ref().is_some_and(has_web_receipt_provenance) {
            return Err(web_receipt_write_refused("try_patch_note_property"));
        }
        reject_existing_secret_gate_property(existing.as_ref(), "try_patch_note_property")?;
        self.inner
            .try_patch_note_property(id, namespace, filter, json_path, value, updated_at)
            .await
    }

    async fn patch_note_property_atomic(
        &self,
        ids: Vec<Uuid>,
        namespace: &str,
        filter: &NoteFilter,
        json_path: &str,
        value: Value,
        updated_at: i64,
    ) -> StorageResult<()> {
        reject_reserved_patch_target(json_path, "patch_note_property_atomic")?;
        // Policy checks include tombstones and keep input order. Hold a secret
        // refusal until every window has been checked for web provenance.
        let mut secret_gate_refusal = None;
        for window in ids.chunks(128) {
            match self.inner.get_notes_batch_including_deleted(window).await {
                Ok(notes) => {
                    let notes: HashMap<Uuid, Note> =
                        notes.into_iter().map(|note| (note.id, note)).collect();
                    for id in window {
                        let existing = notes.get(id);
                        if existing.is_some_and(has_web_receipt_provenance) {
                            return Err(web_receipt_write_refused("patch_note_property_atomic"));
                        }
                        if secret_gate_refusal.is_none() {
                            secret_gate_refusal = reject_existing_secret_gate_property(
                                existing,
                                "patch_note_property_atomic",
                            )
                            .err();
                        }
                    }
                }
                Err(_) => {
                    // A later malformed row can fail batch decoding before an
                    // earlier policy refusal. Replay this window in input order.
                    for id in window {
                        let existing = self.inner.get_note_including_deleted(*id).await?;
                        let existing = existing.as_ref();
                        if existing.is_some_and(has_web_receipt_provenance) {
                            return Err(web_receipt_write_refused("patch_note_property_atomic"));
                        }
                        if secret_gate_refusal.is_none() {
                            secret_gate_refusal = reject_existing_secret_gate_property(
                                existing,
                                "patch_note_property_atomic",
                            )
                            .err();
                        }
                    }
                }
            }
        }
        if let Some(refusal) = secret_gate_refusal {
            return Err(refusal);
        }
        self.inner
            .patch_note_property_atomic(ids, namespace, filter, json_path, value, updated_at)
            .await
    }

    async fn query_notes(
        &self,
        namespace: &str,
        kind: Option<&str>,
        page: PageRequest,
    ) -> StorageResult<Page<Note>> {
        self.inner.query_notes(namespace, kind, page).await
    }

    async fn query_notes_count_free(
        &self,
        namespace: &str,
        kind: Option<&str>,
        page: PageRequest,
    ) -> StorageResult<Page<Note>> {
        self.inner
            .query_notes_count_free(namespace, kind, page)
            .await
    }

    async fn query_notes_filtered(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        page: PageRequest,
    ) -> StorageResult<Page<Note>> {
        self.inner
            .query_notes_filtered(namespace, filter, page)
            .await
    }

    async fn query_notes_filtered_count_free(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        page: PageRequest,
    ) -> StorageResult<Page<Note>> {
        self.inner
            .query_notes_filtered_count_free(namespace, filter, page)
            .await
    }

    async fn count_notes_filtered_in_snapshot(
        &self,
        namespace: &str,
        filters: &[NoteFilter],
    ) -> StorageResult<Vec<u64>> {
        self.inner
            .count_notes_filtered_in_snapshot(namespace, filters)
            .await
    }

    async fn count_notes_filtered_bounded_in_snapshot(
        &self,
        namespace: &str,
        filters: &[NoteFilter],
        cap: u32,
    ) -> StorageResult<Vec<BoundedCount>> {
        self.inner
            .count_notes_filtered_bounded_in_snapshot(namespace, filters, cap)
            .await
    }

    async fn note_sequence(&self, id: Uuid) -> StorageResult<Option<i64>> {
        self.inner.note_sequence(id).await
    }

    async fn query_notes_filtered_after(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        after: Option<SeekCursor>,
        limit: u32,
    ) -> StorageResult<SeekPage<Note>> {
        self.inner
            .query_notes_filtered_after(namespace, filter, after, limit)
            .await
    }

    async fn query_notes_filtered_bounded(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        max_rows: u32,
    ) -> StorageResult<Vec<Note>> {
        self.inner
            .query_notes_filtered_bounded(namespace, filter, max_rows)
            .await
    }

    async fn count_notes(&self, namespace: &str, kind: Option<&str>) -> StorageResult<u64> {
        self.inner.count_notes(namespace, kind).await
    }

    async fn count_notes_in_namespaces(
        &self,
        namespaces: &[String],
        kind: Option<&str>,
    ) -> StorageResult<u64> {
        self.inner.count_notes_in_namespaces(namespaces, kind).await
    }

    async fn try_insert_note(&self, note: Note) -> StorageResult<bool> {
        reject_reserved_note_properties(note.properties.as_ref(), "try_insert_note")?;
        reject_if_forged_message_note(&note, "try_insert_note")?;
        reject_forged_web_receipt_note(&note, "try_insert_note")?;
        self.inner.try_insert_note(note).await
    }

    async fn get_notes_batch(&self, ids: &[Uuid]) -> StorageResult<Vec<Note>> {
        self.inner.get_notes_batch(ids).await
    }

    async fn get_notes_batch_including_deleted(&self, ids: &[Uuid]) -> StorageResult<Vec<Note>> {
        self.inner.get_notes_batch_including_deleted(ids).await
    }

    async fn get_note_visibility_batch(&self, ids: &[Uuid]) -> StorageResult<Vec<NoteVisibility>> {
        self.inner.get_note_visibility_batch(ids).await
    }
}

#[cfg(test)]
mod tests;