osproxy-sink 1.0.1

//! The [`Reader`] trait: fetching a single document by physical id.
//!
//! Reads are always direct-to-cluster, unlike writes, they cannot be served by
//! a queue, so the read seam is separate from [`Sink`](crate::Sink). The same
//! backend type may implement both (`OpenSearchSink` does, sharing its pooled
//! connection), while a write-only `QueueSink` implements only [`Sink`].
//!
//! [`Sink`]: crate::Sink
//
// JUSTIFY(file-length): one cohesive family of read-path value types, the
// `Reader` trait plus the op (`ReadOp`/`SearchOp`/`CursorOp`) and outcome
// (`ReadOutcome`/`SearchOutcome`/`CountOutcome`/`CursorOutcome`) structs they
// exchange. They share the same builders and conventions; splitting them would
// scatter one small vocabulary across files for no real separation.

use osproxy_core::{ClusterId, Target, TraceContext};
use osproxy_spi::{HttpMethod, Protocol};

use crate::error::SinkError;

/// A read-by-id operation against a resolved [`Target`].
///
/// The id is already the **physical** id (the tenancy adapter mapped the
/// client's logical id, `docs/04` §5); the reader does no rewriting.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct ReadOp {
    /// The physical destination to read from.
    pub target: Target,
    /// The physical document id to fetch.
    pub id: String,
    /// The `_routing` value (the partition id), if the placement routes.
    pub routing: Option<String>,
    /// The upstream wire protocol this read is dispatched over. Defaults to
    /// [`Protocol::Http1`].
    pub protocol: Protocol,
    /// The W3C trace context to forward downstream (`traceparent`), so the
    /// upstream's spans join this request's distributed trace. `None` = no
    /// propagation header is sent.
    pub trace: Option<TraceContext>,
    /// Client headers to relay verbatim to the upstream (the forwarding policy's
    /// output). Applied before [`trace`](Self::trace). Empty by default.
    pub forward_headers: Vec<(String, String)>,
}

impl ReadOp {
    /// Constructs a read operation (defaulting to HTTP/1.1 upstream).
    #[must_use]
    pub fn new(target: Target, id: impl Into<String>, routing: Option<String>) -> Self {
        Self {
            target,
            id: id.into(),
            routing,
            protocol: Protocol::Http1,
            trace: None,
            forward_headers: Vec::new(),
        }
    }

    /// Sets the upstream protocol for this op (builder style).
    #[must_use]
    pub fn with_protocol(mut self, protocol: Protocol) -> Self {
        self.protocol = protocol;
        self
    }

    /// Sets the trace context to propagate downstream (builder style).
    #[must_use]
    pub fn with_trace(mut self, trace: Option<TraceContext>) -> Self {
        self.trace = trace;
        self
    }

    /// Sets the client headers to relay verbatim to the upstream (builder style).
    #[must_use]
    pub fn with_forward_headers(mut self, headers: Vec<(String, String)>) -> Self {
        self.forward_headers = headers;
        self
    }
}

/// The outcome of a read: whether the document was found, and its raw upstream
/// body (the document as stored, before the read-path field strip).
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct ReadOutcome {
    /// The upstream HTTP status.
    pub status: u16,
    /// Whether the document exists.
    pub found: bool,
    /// The raw upstream response body (the stored document when `found`).
    pub body: Vec<u8>,
    /// Whether this read rode a reused pooled connection (NFR-P telemetry).
    pub pool_reuse: bool,
}

impl ReadOutcome {
    /// A hit carrying the stored document body.
    #[must_use]
    pub fn found(status: u16, body: Vec<u8>) -> Self {
        Self {
            status,
            found: true,
            body,
            pool_reuse: false,
        }
    }

    /// A miss (no such document).
    #[must_use]
    pub fn not_found(status: u16, body: Vec<u8>) -> Self {
        Self {
            status,
            found: false,
            body,
            pool_reuse: false,
        }
    }

    /// Records whether the dispatch reused a pooled connection (builder style).
    #[must_use]
    pub fn with_pool_reuse(mut self, reused: bool) -> Self {
        self.pool_reuse = reused;
        self
    }
}

/// A search operation against a resolved [`Target`].
///
/// The body is the **already-wrapped** query (the tenancy partition filter has
/// been applied, `docs/04` §4); the reader forwards it verbatim.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct SearchOp {
    /// The physical destination to search.
    pub target: Target,
    /// The query body to forward upstream (already partition-filtered).
    pub body: Vec<u8>,
    /// The upstream wire protocol this search is dispatched over. Defaults to
    /// [`Protocol::Http1`].
    pub protocol: Protocol,
    /// An already-allow-listed query string (without the `?`) to append to the
    /// upstream URL, e.g. `scroll=1m` to open a scroll. The engine filters this
    /// to cursor-safe params before it reaches here; the sink appends it verbatim.
    pub query: Option<String>,
    /// The W3C trace context to forward downstream (`traceparent`).
    pub trace: Option<TraceContext>,
    /// Client headers to relay verbatim to the upstream (the forwarding policy's
    /// output). Applied before [`trace`](Self::trace). Empty by default.
    pub forward_headers: Vec<(String, String)>,
}

impl SearchOp {
    /// Constructs a search operation (defaulting to HTTP/1.1 upstream).
    #[must_use]
    pub fn new(target: Target, body: Vec<u8>) -> Self {
        Self {
            target,
            body,
            protocol: Protocol::Http1,
            query: None,
            trace: None,
            forward_headers: Vec::new(),
        }
    }

    /// Sets the upstream protocol for this op (builder style).
    #[must_use]
    pub fn with_protocol(mut self, protocol: Protocol) -> Self {
        self.protocol = protocol;
        self
    }

    /// Sets the (already allow-listed) upstream query string (builder style).
    #[must_use]
    pub fn with_query(mut self, query: Option<String>) -> Self {
        self.query = query;
        self
    }

    /// Sets the trace context to propagate downstream (builder style).
    #[must_use]
    pub fn with_trace(mut self, trace: Option<TraceContext>) -> Self {
        self.trace = trace;
        self
    }

    /// Sets the client headers to relay verbatim to the upstream (builder style).
    #[must_use]
    pub fn with_forward_headers(mut self, headers: Vec<(String, String)>) -> Self {
        self.forward_headers = headers;
        self
    }
}

/// The outcome of a search: the upstream status and raw response body (the
/// hits, before the read-path field strip).
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct SearchOutcome {
    /// The upstream HTTP status.
    pub status: u16,
    /// The raw upstream response body (the hits envelope).
    pub body: Vec<u8>,
    /// Whether this search rode a reused pooled connection (NFR-P telemetry).
    pub pool_reuse: bool,
}

impl SearchOutcome {
    /// Constructs a search outcome.
    #[must_use]
    pub fn new(status: u16, body: Vec<u8>) -> Self {
        Self {
            status,
            body,
            pool_reuse: false,
        }
    }

    /// Records whether the dispatch reused a pooled connection (builder style).
    #[must_use]
    pub fn with_pool_reuse(mut self, reused: bool) -> Self {
        self.pool_reuse = reused;
        self
    }
}

/// The outcome of a count: the upstream status and the matched document count.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct CountOutcome {
    /// The upstream HTTP status.
    pub status: u16,
    /// The number of matching documents.
    pub count: u64,
    /// Whether this count rode a reused pooled connection (NFR-P telemetry).
    pub pool_reuse: bool,
}

impl CountOutcome {
    /// Constructs a count outcome.
    #[must_use]
    pub fn new(status: u16, count: u64) -> Self {
        Self {
            status,
            count,
            pool_reuse: false,
        }
    }

    /// Records whether the dispatch reused a pooled connection (builder style).
    #[must_use]
    pub fn with_pool_reuse(mut self, reused: bool) -> Self {
        self.pool_reuse = reused;
        self
    }
}

/// A raw cursor passthrough op (`docs/03` §6): forward `method path` with `body`
/// to the specific `cluster` the cursor is pinned to, scroll/PIT continue,
/// clear, or close. Unlike the typed ops, the destination is *already resolved*
/// (the engine recovered it from the cursor's signed envelope), so this carries
/// the cluster directly rather than a partition.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct CursorOp {
    /// The cluster the cursor is pinned to.
    pub cluster: ClusterId,
    /// The HTTP method to forward (continue is `POST`/`GET`, clear/close `DELETE`).
    pub method: HttpMethod,
    /// The upstream path (e.g. `/_search/scroll`), already with the real cursor id.
    pub path: String,
    /// The request body to forward (the real, unwrapped cursor id substituted in).
    pub body: Vec<u8>,
    /// An already-allow-listed query string (without the `?`) to append to the
    /// upstream URL, e.g. `keep_alive=1m` on PIT create. Filtered by the engine.
    pub query: Option<String>,
    /// The pinned cluster's base URL, when the engine knows it (the placement that
    /// opened the cursor supplied it). `None` for an affinity continue recovered
    /// from the envelope alone: the sink then reuses the pool the opening request
    /// already built for this cluster, erroring only if none exists.
    pub endpoint: Option<String>,
    /// The upstream wire protocol. Defaults to [`Protocol::Http1`].
    pub protocol: Protocol,
    /// The W3C trace context to forward downstream.
    pub trace: Option<TraceContext>,
    /// Client headers to relay verbatim to the upstream (the forwarding policy's
    /// output). Applied before [`trace`](Self::trace), so a proxy-injected trace
    /// header wins when span export is on. Empty by default.
    pub forward_headers: Vec<(String, String)>,
}

impl CursorOp {
    /// Constructs a cursor passthrough op (defaulting to HTTP/1.1 upstream).
    #[must_use]
    pub fn new(
        cluster: ClusterId,
        method: HttpMethod,
        path: impl Into<String>,
        body: Vec<u8>,
    ) -> Self {
        Self {
            cluster,
            method,
            path: path.into(),
            body,
            query: None,
            endpoint: None,
            protocol: Protocol::Http1,
            trace: None,
            forward_headers: Vec::new(),
        }
    }

    /// Sets the client headers to relay verbatim to the upstream (builder style).
    #[must_use]
    pub fn with_forward_headers(mut self, headers: Vec<(String, String)>) -> Self {
        self.forward_headers = headers;
        self
    }

    /// Sets the pinned cluster's base URL (builder style), when the engine knows
    /// it. Lets the sink build the pool for an affinity request even on an
    /// instance that did not serve the opening call.
    #[must_use]
    pub fn with_endpoint(mut self, endpoint: Option<String>) -> Self {
        self.endpoint = endpoint;
        self
    }

    /// Sets the upstream wire protocol (builder style).
    #[must_use]
    pub fn with_protocol(mut self, protocol: Protocol) -> Self {
        self.protocol = protocol;
        self
    }

    /// Sets the (already allow-listed) upstream query string (builder style).
    #[must_use]
    pub fn with_query(mut self, query: Option<String>) -> Self {
        self.query = query;
        self
    }

    /// Sets the trace context to propagate downstream (builder style).
    #[must_use]
    pub fn with_trace(mut self, trace: Option<TraceContext>) -> Self {
        self.trace = trace;
        self
    }
}

/// A verbatim forward whose request body is a **stream**, not buffered bytes
/// (ADR-014 stage 2): the same destination shape as [`CursorOp`] but the body is
/// supplied separately as a [`ByteBody`](crate::ByteBody) so it can be
/// piped from the downstream connection straight to the upstream without ever
/// being collected. Used by the tenant-agnostic passthrough path.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct ForwardOp {
    /// The cluster to forward to.
    pub cluster: ClusterId,
    /// The HTTP method to forward.
    pub method: HttpMethod,
    /// The upstream path, forwarded verbatim.
    pub path: String,
    /// An already-allow-listed query string (without the `?`) to append upstream.
    pub query: Option<String>,
    /// The cluster's base URL, when known (so the pool can be built on any
    /// instance). `None` reuses an existing pool, erroring if none exists.
    pub endpoint: Option<String>,
    /// The upstream wire protocol. Defaults to [`Protocol::Http1`].
    pub protocol: Protocol,
    /// The W3C trace context to forward downstream.
    pub trace: Option<TraceContext>,
    /// Client headers to relay verbatim to the upstream (the forwarding policy's
    /// output). Applied before [`trace`](Self::trace). Empty by default.
    pub forward_headers: Vec<(String, String)>,
}

impl ForwardOp {
    /// Constructs a streaming forward op (defaulting to HTTP/1.1 upstream).
    #[must_use]
    pub fn new(cluster: ClusterId, method: HttpMethod, path: impl Into<String>) -> Self {
        Self {
            cluster,
            method,
            path: path.into(),
            query: None,
            endpoint: None,
            protocol: Protocol::Http1,
            trace: None,
            forward_headers: Vec::new(),
        }
    }

    /// Sets the client headers to relay verbatim to the upstream (builder style).
    #[must_use]
    pub fn with_forward_headers(mut self, headers: Vec<(String, String)>) -> Self {
        self.forward_headers = headers;
        self
    }

    /// Sets the cluster's base URL (builder style).
    #[must_use]
    pub fn with_endpoint(mut self, endpoint: Option<String>) -> Self {
        self.endpoint = endpoint;
        self
    }

    /// Sets the (already allow-listed) upstream query string (builder style).
    #[must_use]
    pub fn with_query(mut self, query: Option<String>) -> Self {
        self.query = query;
        self
    }

    /// Sets the upstream wire protocol (builder style).
    #[must_use]
    pub fn with_protocol(mut self, protocol: Protocol) -> Self {
        self.protocol = protocol;
        self
    }

    /// Sets the trace context to propagate downstream (builder style).
    #[must_use]
    pub fn with_trace(mut self, trace: Option<TraceContext>) -> Self {
        self.trace = trace;
        self
    }
}

/// The outcome of a **streaming** verbatim forward (ADR-014): the upstream status
/// and its response body as a live [`ByteBody`](crate::ByteBody) stream, piped
/// back to the client without ever being collected. Unlike [`CursorOutcome`], the
/// body is not materialized here, so this carries no derives (the stream is
/// one-shot).
pub struct StreamingForward {
    /// The upstream HTTP status.
    pub status: u16,
    /// The upstream response body, streamed back verbatim.
    pub body: crate::ByteBody,
    /// The upstream `Content-Type`, forwarded verbatim so a non-JSON passthrough
    /// body is not mislabeled `application/json`. `None` ⇒ the caller defaults to
    /// JSON.
    pub content_type: Option<String>,
    /// Whether this op rode a reused pooled connection (NFR-P telemetry).
    pub pool_reuse: bool,
}

impl std::fmt::Debug for StreamingForward {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // The streamed body is not `Debug`; show the shape.
        f.debug_struct("StreamingForward")
            .field("status", &self.status)
            .field("pool_reuse", &self.pool_reuse)
            .finish_non_exhaustive()
    }
}

/// The outcome of a **streaming** search (ADR-014, final stage): the upstream
/// status and its response body as a live [`ByteBody`](crate::ByteBody), piped
/// back through the engine's hit transform without ever being collected. Like
/// [`StreamingForward`], the body is one-shot, so this carries no derives.
pub struct StreamingSearch {
    /// The upstream HTTP status.
    pub status: u16,
    /// The upstream response body, streamed back to be transformed on the fly.
    pub body: crate::ByteBody,
    /// Whether this search rode a reused pooled connection (NFR-P telemetry).
    pub pool_reuse: bool,
}

impl std::fmt::Debug for StreamingSearch {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StreamingSearch")
            .field("status", &self.status)
            .field("pool_reuse", &self.pool_reuse)
            .finish_non_exhaustive()
    }
}

/// The outcome of a cursor passthrough: the upstream status and raw body,
/// forwarded back to the client verbatim.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct CursorOutcome {
    /// The upstream HTTP status.
    pub status: u16,
    /// The raw upstream response body.
    pub body: Vec<u8>,
    /// The upstream `Content-Type`, forwarded verbatim so an admin/cursor
    /// passthrough body (e.g. a `_cat` `text/plain`) is not mislabeled
    /// `application/json`. `None` ⇒ the caller defaults to JSON.
    pub content_type: Option<String>,
    /// Whether this op rode a reused pooled connection (NFR-P telemetry).
    pub pool_reuse: bool,
}

impl CursorOutcome {
    /// Constructs a cursor outcome.
    #[must_use]
    pub fn new(status: u16, body: Vec<u8>) -> Self {
        Self {
            status,
            body,
            content_type: None,
            pool_reuse: false,
        }
    }

    /// Records whether the dispatch reused a pooled connection (builder style).
    #[must_use]
    pub fn with_pool_reuse(mut self, reused: bool) -> Self {
        self.pool_reuse = reused;
        self
    }

    /// Carries the upstream `Content-Type` (builder style).
    #[must_use]
    pub fn with_content_type(mut self, content_type: Option<String>) -> Self {
        self.content_type = content_type;
        self
    }
}

/// Where reads come from.
///
/// The read counterpart of [`Sink`](crate::Sink). Kept separate because a read
/// is inherently direct-to-cluster: a redundancy `QueueSink` can absorb writes
/// but cannot answer a get-by-id or a search.
///
/// # Invariants
///
/// - MUST NOT panic; return [`SinkError`] for every transport/upstream failure
///   (NFR-R1). A missing document is *not* an error, it is a
///   [`ReadOutcome`] with `found == false`.
pub trait Reader: Send + Sync {
    /// Fetches a single document by physical id.
    ///
    /// # Errors
    ///
    /// Returns [`SinkError`] if the upstream cannot be reached or returns a
    /// server error (a 404 for a missing document is a normal not-found
    /// outcome, not an error).
    fn get(
        &self,
        op: ReadOp,
    ) -> impl std::future::Future<Output = Result<ReadOutcome, SinkError>> + Send;

    /// Runs a search, returning the raw hits envelope.
    ///
    /// # Errors
    ///
    /// Returns [`SinkError`] if the upstream cannot be reached or returns a
    /// server error.
    fn search(
        &self,
        op: SearchOp,
    ) -> impl std::future::Future<Output = Result<SearchOutcome, SinkError>> + Send;

    /// Counts the documents matching a (partition-filtered) query.
    ///
    /// Takes the same [`SearchOp`] as [`Reader::search`], the wrapped query is
    /// identical, but hits the count endpoint, returning only the total.
    ///
    /// # Errors
    ///
    /// Returns [`SinkError`] if the upstream cannot be reached or returns a
    /// server error.
    fn count(
        &self,
        op: SearchOp,
    ) -> impl std::future::Future<Output = Result<CountOutcome, SinkError>> + Send;

    /// Forwards a raw cursor request to its pinned cluster (scroll/PIT continue,
    /// clear, close). The default is **unsupported**, a sink that cannot
    /// passthrough (the in-memory test sink, a write-only queue) rejects it;
    /// `OpenSearchSink` overrides it with a real upstream call.
    ///
    /// # Errors
    ///
    /// Returns [`SinkError`] if the sink does not support passthrough or the
    /// upstream cannot be reached.
    fn cursor(
        &self,
        _op: CursorOp,
    ) -> impl std::future::Future<Output = Result<CursorOutcome, SinkError>> + Send {
        async {
            Err(SinkError::Transport {
                kind: "cursor passthrough not supported by this sink",
            })
        }
    }

    /// Runs a search whose **response** streams back (ADR-014, final stage): the
    /// upstream hits envelope is piped to the engine's hit transform without being
    /// collected, so a large response (e.g. heavy `aggregations`) never lands in
    /// memory. The default is **unsupported**; `OpenSearchSink` overrides it with a
    /// real streamed upstream call.
    ///
    /// # Errors
    ///
    /// Returns [`SinkError`] if the sink does not support streaming search or the
    /// upstream cannot be reached or returns a server error.
    fn search_stream(
        &self,
        _op: SearchOp,
    ) -> impl std::future::Future<Output = Result<StreamingSearch, SinkError>> + Send {
        async {
            Err(SinkError::Transport {
                kind: "streaming search not supported by this sink",
            })
        }
    }

    /// Forwards a request to a cluster with the body supplied as a **stream**
    /// (ADR-014 stage 2): the verbatim-passthrough path pipes the downstream body
    /// straight upstream without buffering. The default is **unsupported**;
    /// `OpenSearchSink` overrides it with a real streamed upstream call.
    ///
    /// # Errors
    ///
    /// Returns [`SinkError`] if the sink does not support streaming forward or the
    /// upstream cannot be reached.
    fn forward_stream(
        &self,
        _op: ForwardOp,
        _body: crate::opensearch::ByteBody,
    ) -> impl std::future::Future<Output = Result<StreamingForward, SinkError>> + Send {
        async {
            Err(SinkError::Transport {
                kind: "streaming forward not supported by this sink",
            })
        }
    }
}