graphddb_runtime 0.7.5

//! Host-side middleware / hooks — the Rust parity counterpart of
//! `python/graphddb_runtime/middleware.py` and the merged PHP `php/src/Middleware/*`.
//!
//! It implements the SAME 10 hook points as the reference runtimes — read R1–R5
//! and write W1–W5 — with the SAME ordering (`before*` FIFO / first-registered
//! first; `after*` / `onError` LIFO / last-registered first), onion-style context
//! threading through `after*`, `onError` recovery, and the same context objects /
//! fields Python and PHP pass. A middleware is NEVER serialized: it lives only on
//! the runtime's host-side registry and never touches `operations.json`.
//!
//! Hook points and their seams:
//! - **R1** `read.before`      — request entry; may mutate [`ReadRequestCtx::params`], may error to cancel.
//! - **R2** `read.op.before`   — one physical op, before send; may mutate [`ReadOpCtx`].
//! - **R3** `read.op.afterFetch` — the op's raw items; onion-threaded, may replace them.
//! - **R4** `read.afterFetch`  — the final assembled result; onion-threaded, may replace it.
//! - **R5** `read.onError`     — request- OR op-level failure; may recover by returning a value.
//! - **W1** `write.before`     — one logical write; may mutate [`WriteCtx::kind`] / [`WriteCtx::input`], may error.
//! - **W2** `write.after`      — after commit, the change; observe-only.
//! - **W3** `write.persist.before` — the composed physical batch; may mutate items, may error.
//! - **W4** `write.persist.after`  — the executor results; observe-only.
//! - **W5** `write.onError`    — logical- OR persist-level failure; may recover.

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

use aws_sdk_dynamodb::types::AttributeValue;
use serde_json::{Map as JsonMap, Value as Json};

use crate::errors::GraphDDBError;

// ── context objects ───────────────────────────────────────────────────────────

/// R1 / R4 / R5 (request-level) context. `before` (R1) may mutate [`Self::params`]
/// and may error to cancel; `afterFetch` (R4) receives the final assembled result
/// and may return a replacement; `onError` (R5) may recover by returning a value.
#[derive(Debug, Clone)]
pub struct ReadRequestCtx {
    /// The request kind (`"query"`).
    pub kind: String,
    /// The `ctx.model` handle (entity name + manifest meta, or null on the read path).
    pub model: Json,
    /// The per-call context threaded from `options.context` (`{}` when none).
    pub context: Json,
    /// The mutable request params (`{"params": {...}, "options": {...}}`), which an
    /// R1 hook may rewrite before key-resolution.
    pub params: JsonMap<String, Json>,
    /// Per-request scratch a hook can stash values in across R1→R4→R5.
    pub state: JsonMap<String, Json>,
}

/// R2 / R3 / R5 (op-level) context — fires for the root read AND every relation
/// fan-out fetch. `before` (R2) may mutate [`Self::operation`] before send;
/// `afterFetch` (R3) may transform the raw items; `onError` (R5, op-level) may
/// recover with an item list.
///
/// The physical operation is carried as a typed request (the seam the conformance
/// harness inspects: consistent-read capture, filter-strip). `op_type` is the
/// `ctx.operation.type` discriminant Python/PHP expose.
#[derive(Debug, Clone)]
pub struct ReadOpCtx {
    /// The op kind: `"GetItem"` | `"Query"` | `"BatchGetItem"`.
    pub op_type: String,
    /// The `ctx.model` handle.
    pub model: Json,
    /// The per-call context.
    pub context: Json,
    /// The relation path (`[]` for the root, the property chain for a fan-out leg).
    pub relation_path: Vec<String>,
    /// Physical table name.
    pub table_name: String,
    /// For a GetItem: the primary key (empty otherwise).
    pub key: HashMap<String, AttributeValue>,
    /// For a GetItem: the strongly-consistent flag.
    pub consistent_read: bool,
    /// `ExpressionAttributeNames` (Query).
    pub names: HashMap<String, String>,
    /// `ExpressionAttributeValues` (Query).
    pub values: HashMap<String, AttributeValue>,
    /// `FilterExpression` (Query), if any.
    pub filter_expression: Option<String>,
    /// Per-op scratch.
    pub state: JsonMap<String, Json>,
}

impl ReadOpCtx {
    /// A GetItem read-op context.
    pub fn get_item(
        table_name: String,
        key: HashMap<String, AttributeValue>,
        consistent_read: bool,
        relation_path: Vec<String>,
        model: Json,
        context: Json,
    ) -> Self {
        Self {
            op_type: "GetItem".to_string(),
            model,
            context,
            relation_path,
            table_name,
            key,
            consistent_read,
            names: HashMap::new(),
            values: HashMap::new(),
            filter_expression: None,
            state: JsonMap::new(),
        }
    }

    /// A Query read-op context.
    pub fn query(
        table_name: String,
        names: HashMap<String, String>,
        values: HashMap<String, AttributeValue>,
        filter_expression: Option<String>,
        relation_path: Vec<String>,
        model: Json,
        context: Json,
    ) -> Self {
        Self {
            op_type: "Query".to_string(),
            model,
            context,
            relation_path,
            table_name,
            key: HashMap::new(),
            consistent_read: false,
            names,
            values,
            filter_expression,
            state: JsonMap::new(),
        }
    }

    /// A BatchGetItem read-op context (belongsTo/hasOne fan-out).
    pub fn batch_get(
        table_name: String,
        names: HashMap<String, String>,
        relation_path: Vec<String>,
        model: Json,
        context: Json,
    ) -> Self {
        Self {
            op_type: "BatchGetItem".to_string(),
            model,
            context,
            relation_path,
            table_name,
            key: HashMap::new(),
            consistent_read: false,
            names,
            values: HashMap::new(),
            filter_expression: None,
            state: JsonMap::new(),
        }
    }
}

/// W1 / W2 / W5 (logical-write-level) context. `before` (W1) may mutate
/// [`Self::kind`] and [`Self::input`] (incl. the delete→update soft-delete rewrite)
/// and may error to cancel (in a transaction, aborting ALL ops); `after` (W2)
/// observes the committed change; `onError` (W5) may recover.
#[derive(Debug, Clone)]
pub struct WriteCtx {
    /// The logical write kind: `"put"` | `"update"` | `"delete"` (W1 may rewrite it).
    pub kind: String,
    /// The `ctx.model` handle (entity name + meta; concrete for writes).
    pub model: Json,
    /// The per-call context.
    pub context: Json,
    /// The mutable write input (`item` for a put; `key` / `changes` for an
    /// update/delete; the whole caller `params` under `params`).
    pub input: JsonMap<String, Json>,
    /// Per-write scratch across W1→W2→W5.
    pub state: JsonMap<String, Json>,
    /// The transaction identity when this op is part of an atomic batch (`None`
    /// for a single-op write). Shared by all logical ops of one transaction.
    pub transaction: Option<u64>,
}

/// W3 / W4 / W5 (physical-persist-level) context — fires AFTER derivation, on the
/// REAL composed batch. `before` (W3) may mutate [`Self::items`], may error to
/// abort; `after` (W4) observes the executor results; `onError` (W5) may recover.
///
/// In a transaction the persist hooks fire ONCE for the whole atomic batch; for a
/// single-op write `items` is a one-element list. Each item is a composed
/// transact-item-shaped mutation whose expression fields a W3 hook may edit.
#[derive(Debug, Clone)]
pub struct PersistCtx {
    /// The composed physical write items (Put/Update/Delete/ConditionCheck).
    pub items: Vec<PersistItemCtx>,
    /// The `{model, kind}` origin of each logical write feeding the batch.
    pub origins: Vec<Json>,
    /// The per-call context.
    pub context: Json,
    /// Per-persist scratch.
    pub state: JsonMap<String, Json>,
    /// The transaction identity, when this persist is an atomic batch.
    pub transaction: Option<u64>,
}

/// One composed physical write item the W3 hook sees — the op kind + its mutable
/// expression fields (mirrors the transact-item `{Put|Update|Delete: body}` shape
/// the PHP hook strips a compiled condition off).
#[derive(Debug, Clone)]
pub struct PersistItemCtx {
    /// The op kind: `"Put"` | `"Update"` | `"Delete"` | `"ConditionCheck"`.
    pub op_kind: String,
    /// `ConditionExpression`, if any.
    pub condition_expression: Option<String>,
    /// `ExpressionAttributeNames`.
    pub names: HashMap<String, String>,
    /// `ExpressionAttributeValues`.
    pub values: HashMap<String, AttributeValue>,
}

impl PersistItemCtx {
    /// A persist item context for a single composed write with only a condition.
    pub fn new(
        op_kind: impl Into<String>,
        condition_expression: Option<String>,
        names: HashMap<String, String>,
        values: HashMap<String, AttributeValue>,
    ) -> Self {
        Self {
            op_kind: op_kind.into(),
            condition_expression,
            names,
            values,
        }
    }
}

// ── onError recovery marker ───────────────────────────────────────────────────

/// An `onError` hook return, interpreting recovery like Python's `_recovered`:
/// `Decline` = did not recover (rethrow); `Recover(value)` = recovered with that
/// value (which may itself be `Json::Null`, the recover-with-none case).
#[derive(Debug, Clone)]
pub enum Recovery {
    /// Did not recover — the original error propagates.
    Decline,
    /// Recovered with this value.
    Recover(Json),
}

// ── hook closure aliases ──────────────────────────────────────────────────────

type ReadBefore = Arc<dyn Fn(&mut ReadRequestCtx) -> Result<(), GraphDDBError> + Send + Sync>;
type ReadAfter = Arc<dyn Fn(&ReadRequestCtx, Json) -> Json + Send + Sync>;
type ReadOnError = Arc<dyn Fn(&ReadRequestCtx, &GraphDDBError) -> Recovery + Send + Sync>;
type OpBefore = Arc<dyn Fn(&mut ReadOpCtx) -> Result<(), GraphDDBError> + Send + Sync>;
type OpAfter = Arc<dyn Fn(&ReadOpCtx, Vec<Json>) -> Vec<Json> + Send + Sync>;
type OpOnError = Arc<dyn Fn(&ReadOpCtx, &GraphDDBError) -> Option<Vec<Json>> + Send + Sync>;
type WriteBefore = Arc<dyn Fn(&mut WriteCtx) -> Result<(), GraphDDBError> + Send + Sync>;
type WriteAfter = Arc<dyn Fn(&WriteCtx, &Json) + Send + Sync>;
type WriteOnError = Arc<dyn Fn(&WriteCtx, &GraphDDBError) -> Recovery + Send + Sync>;
type PersistBefore = Arc<dyn Fn(&mut PersistCtx) -> Result<(), GraphDDBError> + Send + Sync>;
type PersistAfter = Arc<dyn Fn(&PersistCtx, &Json) + Send + Sync>;
type PersistOnError = Arc<dyn Fn(&PersistCtx, &GraphDDBError) -> Recovery + Send + Sync>;

/// A registered set of hooks. A host builds this with the builder methods and
/// passes it to [`crate::runtime::GraphDDBRuntime::use_middleware`]. Any subset of
/// the 10 hook points may be set; unset points are simply skipped.
#[derive(Default, Clone)]
pub struct Middleware {
    pub(crate) read_before: Option<ReadBefore>,
    pub(crate) read_after: Option<ReadAfter>,
    pub(crate) read_on_error: Option<ReadOnError>,
    pub(crate) op_before: Option<OpBefore>,
    pub(crate) op_after: Option<OpAfter>,
    pub(crate) op_on_error: Option<OpOnError>,
    pub(crate) write_before: Option<WriteBefore>,
    pub(crate) write_after: Option<WriteAfter>,
    pub(crate) write_on_error: Option<WriteOnError>,
    pub(crate) persist_before: Option<PersistBefore>,
    pub(crate) persist_after: Option<PersistAfter>,
    pub(crate) persist_on_error: Option<PersistOnError>,
}

impl Middleware {
    /// A fresh empty middleware.
    pub fn new() -> Self {
        Self::default()
    }

    /// Register R1 `read.before` (request entry; may mutate params, may error).
    pub fn read_before(
        mut self,
        f: impl Fn(&mut ReadRequestCtx) -> Result<(), GraphDDBError> + Send + Sync + 'static,
    ) -> Self {
        self.read_before = Some(Arc::new(f));
        self
    }
    /// Register R4 `read.afterFetch` (final result; onion-threaded, may replace).
    pub fn read_after(
        mut self,
        f: impl Fn(&ReadRequestCtx, Json) -> Json + Send + Sync + 'static,
    ) -> Self {
        self.read_after = Some(Arc::new(f));
        self
    }
    /// Register R5 `read.onError` (request-level recovery).
    pub fn read_on_error(
        mut self,
        f: impl Fn(&ReadRequestCtx, &GraphDDBError) -> Recovery + Send + Sync + 'static,
    ) -> Self {
        self.read_on_error = Some(Arc::new(f));
        self
    }
    /// Register R2 `read.op.before` (physical op; may mutate the op, may error).
    pub fn read_op_before(
        mut self,
        f: impl Fn(&mut ReadOpCtx) -> Result<(), GraphDDBError> + Send + Sync + 'static,
    ) -> Self {
        self.op_before = Some(Arc::new(f));
        self
    }
    /// Register R3 `read.op.afterFetch` (raw items; onion-threaded, may replace).
    pub fn read_op_after(
        mut self,
        f: impl Fn(&ReadOpCtx, Vec<Json>) -> Vec<Json> + Send + Sync + 'static,
    ) -> Self {
        self.op_after = Some(Arc::new(f));
        self
    }
    /// Register R5 (op-level) `read.op.onError` (recover with an item list).
    pub fn read_op_on_error(
        mut self,
        f: impl Fn(&ReadOpCtx, &GraphDDBError) -> Option<Vec<Json>> + Send + Sync + 'static,
    ) -> Self {
        self.op_on_error = Some(Arc::new(f));
        self
    }
    /// Register W1 `write.before` (logical write; may mutate kind/input, may error).
    pub fn write_before(
        mut self,
        f: impl Fn(&mut WriteCtx) -> Result<(), GraphDDBError> + Send + Sync + 'static,
    ) -> Self {
        self.write_before = Some(Arc::new(f));
        self
    }
    /// Register W2 `write.after` (committed change; observe-only).
    pub fn write_after(mut self, f: impl Fn(&WriteCtx, &Json) + Send + Sync + 'static) -> Self {
        self.write_after = Some(Arc::new(f));
        self
    }
    /// Register W5 (logical-level) `write.onError` (recovery).
    pub fn write_on_error(
        mut self,
        f: impl Fn(&WriteCtx, &GraphDDBError) -> Recovery + Send + Sync + 'static,
    ) -> Self {
        self.write_on_error = Some(Arc::new(f));
        self
    }
    /// Register W3 `write.persist.before` (composed batch; may mutate items, may error).
    pub fn persist_before(
        mut self,
        f: impl Fn(&mut PersistCtx) -> Result<(), GraphDDBError> + Send + Sync + 'static,
    ) -> Self {
        self.persist_before = Some(Arc::new(f));
        self
    }
    /// Register W4 `write.persist.after` (executor results; observe-only).
    pub fn persist_after(mut self, f: impl Fn(&PersistCtx, &Json) + Send + Sync + 'static) -> Self {
        self.persist_after = Some(Arc::new(f));
        self
    }
    /// Register W5 (persist-level) `write.persist.onError` (recovery).
    pub fn persist_on_error(
        mut self,
        f: impl Fn(&PersistCtx, &GraphDDBError) -> Recovery + Send + Sync + 'static,
    ) -> Self {
        self.persist_on_error = Some(Arc::new(f));
        self
    }
}

/// Back-compat alias: the previous single-hook builder name. Prefer [`Middleware`].
pub type WriteHooks = Middleware;

// ── registry ──────────────────────────────────────────────────────────────────

/// The runtime-scoped, registration-ordered middleware registry (the Rust
/// counterpart of the Python `MiddlewareRegistry` / TS `ClientManager`): append via
/// [`Self::use_hooks`], `before*` run FIFO, `after*` / `onError` run LIFO.
#[derive(Default, Clone)]
pub struct MiddlewareRegistry {
    list: Vec<Middleware>,
}

impl MiddlewareRegistry {
    /// A fresh empty registry.
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a middleware (appended last → runs last in FIFO `before*`).
    pub fn use_hooks(&mut self, mw: Middleware) {
        self.list.push(mw);
    }

    /// Remove all registered middleware.
    pub fn clear(&mut self) {
        self.list.clear();
    }

    /// True when at least one middleware is registered.
    pub fn active(&self) -> bool {
        !self.list.is_empty()
    }

    /// True when at least one middleware sets a W2 `write.after` hook (so the
    /// single-op write path knows to request the `ALL_OLD` pre-write image).
    pub fn has_write_after(&self) -> bool {
        self.list.iter().any(|mw| mw.write_after.is_some())
    }

    // ── R1 / R4 / R5 (request-level) ──

    /// R1 — every `read.before` FIFO; a hook may mutate `ctx.params`, may error.
    pub fn run_request_before(&self, ctx: &mut ReadRequestCtx) -> Result<(), GraphDDBError> {
        for mw in &self.list {
            if let Some(h) = &mw.read_before {
                h(ctx)?;
            }
        }
        Ok(())
    }

    /// R4 — every `read.afterFetch` LIFO, onion-threading the result.
    pub fn run_request_after(&self, ctx: &ReadRequestCtx, result: Json) -> Json {
        let mut current = result;
        for mw in self.list.iter().rev() {
            if let Some(h) = &mw.read_after {
                current = h(ctx, current);
            }
        }
        current
    }

    /// R5 (request-level) — every `read.onError` LIFO; first recover wins, else the
    /// original error propagates (`Err`).
    pub fn run_request_error(
        &self,
        ctx: &ReadRequestCtx,
        err: GraphDDBError,
    ) -> Result<Json, GraphDDBError> {
        for mw in self.list.iter().rev() {
            if let Some(h) = &mw.read_on_error {
                if let Recovery::Recover(v) = h(ctx, &err) {
                    return Ok(v);
                }
            }
        }
        Err(err)
    }

    // ── R2 / R3 / R5 (op-level) ──

    /// R2 — every `read.op.before` FIFO; a hook may mutate `ctx.operation`, may error.
    pub fn run_op_before(&self, ctx: &mut ReadOpCtx) -> Result<(), GraphDDBError> {
        for mw in &self.list {
            if let Some(h) = &mw.op_before {
                h(ctx)?;
            }
        }
        Ok(())
    }

    /// R3 — every `read.op.afterFetch` LIFO, onion-threading the raw items.
    pub fn run_op_after(&self, ctx: &ReadOpCtx, items: Vec<Json>) -> Vec<Json> {
        let mut current = items;
        for mw in self.list.iter().rev() {
            if let Some(h) = &mw.op_after {
                current = h(ctx, current);
            }
        }
        current
    }

    /// R5 (op-level) — every `read.op.onError` LIFO; a hook may recover with an
    /// item list. Returns `Ok(items)` on recovery, `Err` if none recovers.
    pub fn run_op_error(
        &self,
        ctx: &ReadOpCtx,
        err: GraphDDBError,
    ) -> Result<Vec<Json>, GraphDDBError> {
        for mw in self.list.iter().rev() {
            if let Some(h) = &mw.op_on_error {
                if let Some(items) = h(ctx, &err) {
                    return Ok(items);
                }
            }
        }
        Err(err)
    }

    // ── W1 / W2 / W5 (logical-write-level) ──

    /// W1 — every `write.before` FIFO; a hook may mutate `ctx.kind` / `ctx.input`,
    /// may error (in a transaction, aborting the whole batch).
    pub fn run_write_before(&self, ctx: &mut WriteCtx) -> Result<(), GraphDDBError> {
        for mw in &self.list {
            if let Some(h) = &mw.write_before {
                h(ctx)?;
            }
        }
        Ok(())
    }

    /// W2 — every `write.after` LIFO with the committed change. Observe-only.
    pub fn run_write_after(&self, ctx: &WriteCtx, change: &Json) {
        for mw in self.list.iter().rev() {
            if let Some(h) = &mw.write_after {
                h(ctx, change);
            }
        }
    }

    /// W5 (logical-level) — every `write.onError` LIFO; first recover wins, else the
    /// error propagates.
    pub fn run_write_error(
        &self,
        ctx: &WriteCtx,
        err: GraphDDBError,
    ) -> Result<Json, GraphDDBError> {
        for mw in self.list.iter().rev() {
            if let Some(h) = &mw.write_on_error {
                if let Recovery::Recover(v) = h(ctx, &err) {
                    return Ok(v);
                }
            }
        }
        Err(err)
    }

    // ── W3 / W4 / W5 (persist-level) ──

    /// W3 — every `write.persist.before` FIFO; a hook may mutate `ctx.items`, may error.
    pub fn run_persist_before(&self, ctx: &mut PersistCtx) -> Result<(), GraphDDBError> {
        for mw in &self.list {
            if let Some(h) = &mw.persist_before {
                h(ctx)?;
            }
        }
        Ok(())
    }

    /// W4 — every `write.persist.after` LIFO with the executor results. Observe-only.
    pub fn run_persist_after(&self, ctx: &PersistCtx, results: &Json) {
        for mw in self.list.iter().rev() {
            if let Some(h) = &mw.persist_after {
                h(ctx, results);
            }
        }
    }

    /// W5 (persist-level) — every `write.persist.onError` LIFO; first recover wins,
    /// else the error propagates.
    pub fn run_persist_error(
        &self,
        ctx: &PersistCtx,
        err: GraphDDBError,
    ) -> Result<Json, GraphDDBError> {
        for mw in self.list.iter().rev() {
            if let Some(h) = &mw.persist_on_error {
                if let Recovery::Recover(v) = h(ctx, &err) {
                    return Ok(v);
                }
            }
        }
        Err(err)
    }
}