klieo-ops-api 3.2.0

use axum::{
    extract::{Query, State},
    Json,
};
use serde::Deserialize;
use std::sync::Arc;

use crate::error::OpsError;
use crate::pagination::clamp_limit;
use crate::state::OpsState;
use crate::types::StreamSummary;

#[derive(Debug, Deserialize)]
pub(crate) struct ListStreamsParams {
    pub(crate) limit: Option<usize>,
    /// Keyset cursor: the `stream_id` of the last summary the client saw.
    /// The next page starts strictly after it. Absent = first page.
    pub(crate) after: Option<String>,
    /// Deprecated and ignored. Retained only so that callers from before the
    /// keyset cursor existed do not silently get page one — when present, a
    /// deprecation warning is logged and the value drives nothing. Use `after`
    /// (the last seen `stream_id`) to page.
    pub(crate) offset: Option<usize>,
}

/// The response is bounded to `limit` streams per call, so the buffer is never
/// materialised whole. To page, the client passes `after=<last stream_id
/// seen>`; the last `stream_id` in the response is the cursor for the next
/// request. An empty response means the scan is exhausted. The legacy `offset`
/// query param is accepted but ignored (a deprecation warning is logged).
pub(crate) async fn list_streams(
    State(state): State<Arc<OpsState>>,
    Query(params): Query<ListStreamsParams>,
) -> Result<Json<Vec<StreamSummary>>, OpsError> {
    let buffer = state
        .resume_buffer
        .as_ref()
        .ok_or(OpsError::StoreNotConfigured)?;

    if params.offset.is_some() {
        tracing::warn!(
            operation = "list_streams",
            "the `offset` query param is deprecated and ignored; use the `after` keyset cursor (pass the last returned stream_id)"
        );
    }

    let limit = clamp_limit(params.limit);
    let page = buffer
        .list_streams_paged(params.after.clone(), limit)
        .await
        .map_err(|e| {
            tracing::error!(operation = "list_streams", detail = %e, "buffer error");
            OpsError::Internal(Box::new(e))
        })?;

    let summaries = page
        .entries
        .into_iter()
        .map(|e| StreamSummary {
            stream_id: e.stream_id,
            is_terminal: e.is_terminal,
            next_id: e.next_id,
            last_write_unix_ms: e.last_write_unix_ms,
        })
        .collect();

    Ok(Json(summaries))
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use axum::{
        body::Body,
        http::{Request, StatusCode},
        routing::get,
        Router,
    };
    use bytes::Bytes;
    use klieo_auth_common::AllowAnonymous;
    use klieo_bus_memory::MemoryBus;
    use klieo_core::bus::{KeyPage, KvEntry, KvStore, Lease, Revision};
    use klieo_core::error::BusError;
    use klieo_core::resume::{KvResumeBuffer, ResumeBuffer};
    use std::sync::Arc;
    use std::time::Duration;
    use tower::ServiceExt;

    /// `KvStore` whose enumeration fails with a backend error, so a
    /// `KvResumeBuffer` built on it surfaces `ResumeError::Backend` ->
    /// `OpsError::Internal` -> HTTP 500 from `list_streams`.
    struct FailingKv;

    #[async_trait]
    impl KvStore for FailingKv {
        async fn get(&self, _b: &str, _k: &str) -> Result<Option<KvEntry>, BusError> {
            Ok(None)
        }

        async fn put(&self, _b: &str, _k: &str, _v: Bytes) -> Result<Revision, BusError> {
            Ok(1)
        }

        async fn cas(
            &self,
            _b: &str,
            _k: &str,
            _v: Bytes,
            _expected: Option<Revision>,
        ) -> Result<Revision, BusError> {
            Ok(1)
        }

        async fn delete(&self, _b: &str, _k: &str) -> Result<(), BusError> {
            Ok(())
        }

        async fn lease(&self, _b: &str, _k: &str, _ttl: Duration) -> Result<Lease, BusError> {
            Err(BusError::Connection("simulated backend failure".into()))
        }

        async fn keys_paginated(
            &self,
            _bucket: &str,
            _cursor: Option<String>,
            _limit: usize,
        ) -> Result<KeyPage, BusError> {
            Err(BusError::Connection("simulated backend failure".into()))
        }
    }

    fn streams_router(state: Arc<OpsState>) -> Router {
        Router::new()
            .route("/streams", get(list_streams))
            .with_state(state)
    }

    #[tokio::test]
    async fn list_streams_no_buffer_returns_501() {
        let state = Arc::new(OpsState {
            authenticator: Arc::new(AllowAnonymous),
            run_log_store: None,
            task_store: None,
            resume_buffer: None,
        });
        let app = streams_router(state);
        let req = Request::builder()
            .uri("/streams")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_IMPLEMENTED);
    }

    #[tokio::test]
    async fn list_streams_buffer_error_returns_500_without_leaking_detail() {
        let buf = Arc::new(KvResumeBuffer::new(Arc::new(FailingKv)));
        let state = Arc::new(OpsState {
            authenticator: Arc::new(AllowAnonymous),
            run_log_store: None,
            task_store: None,
            resume_buffer: Some(buf),
        });
        let app = streams_router(state);
        let req = Request::builder()
            .uri("/streams")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);

        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let text = String::from_utf8(body.to_vec()).unwrap();
        assert!(
            !text.contains("simulated backend failure"),
            "internal error detail must not leak to the client; got: {text}"
        );
        assert!(
            !text.to_lowercase().contains("connection"),
            "internal error class must not leak to the client; got: {text}"
        );
    }

    #[tokio::test]
    async fn list_streams_returns_all_known_streams() {
        let bus = MemoryBus::new();
        let buf = Arc::new(KvResumeBuffer::new(bus.kv.clone()));
        buf.record("s-1", 1, Bytes::from_static(b"x"))
            .await
            .unwrap();
        buf.record("s-2", 1, Bytes::from_static(b"x"))
            .await
            .unwrap();
        buf.close("s-2").await.unwrap();

        let state = Arc::new(OpsState {
            authenticator: Arc::new(AllowAnonymous),
            run_log_store: None,
            task_store: None,
            resume_buffer: Some(buf),
        });
        let app = streams_router(state);
        let req = Request::builder()
            .uri("/streams")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let streams: Vec<serde_json::Value> = serde_json::from_slice(&body).unwrap();
        assert_eq!(streams.len(), 2);
        let s2 = streams.iter().find(|s| s["stream_id"] == "s-2").unwrap();
        assert_eq!(s2["is_terminal"], true);
    }

    async fn stream_ids_for_uri(app: Router, uri: &str) -> Vec<String> {
        let req = Request::builder().uri(uri).body(Body::empty()).unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let streams: Vec<serde_json::Value> = serde_json::from_slice(&body).unwrap();
        streams
            .iter()
            .map(|s| s["stream_id"].as_str().unwrap().to_string())
            .collect()
    }

    #[tokio::test]
    async fn list_streams_limit_caps_the_first_page() {
        let bus = MemoryBus::new();
        let buf = Arc::new(KvResumeBuffer::new(bus.kv.clone()));
        for index in 0..5 {
            buf.record(&format!("s-{index}"), 1, Bytes::from_static(b"x"))
                .await
                .unwrap();
        }

        let state = Arc::new(OpsState {
            authenticator: Arc::new(AllowAnonymous),
            run_log_store: None,
            task_store: None,
            resume_buffer: Some(buf),
        });
        let app = streams_router(state);

        // keys_paginated returns ascending order, so the full list is sorted;
        // derive the expected first two from it rather than hardcoding order.
        let full = stream_ids_for_uri(app.clone(), "/streams?limit=200").await;
        assert_eq!(full.len(), 5);
        let expected_first: Vec<String> = full.iter().take(2).cloned().collect();

        let page = stream_ids_for_uri(app, "/streams?limit=2").await;
        assert_eq!(page.len(), 2, "limit caps the first page to two streams");
        assert_eq!(page, expected_first, "first page is the ascending prefix");
        // Negative fixture: a stream beyond the requested page must be absent.
        assert!(
            !page.contains(&full[2]),
            "stream after the page must not appear"
        );
    }

    #[tokio::test]
    async fn list_streams_after_cursor_returns_the_next_page() {
        let bus = MemoryBus::new();
        let buf = Arc::new(KvResumeBuffer::new(bus.kv.clone()));
        for index in 0..5 {
            buf.record(&format!("s-{index}"), 1, Bytes::from_static(b"x"))
                .await
                .unwrap();
        }

        let state = Arc::new(OpsState {
            authenticator: Arc::new(AllowAnonymous),
            run_log_store: None,
            task_store: None,
            resume_buffer: Some(buf),
        });
        let app = streams_router(state);

        let full = stream_ids_for_uri(app.clone(), "/streams?limit=200").await;
        assert_eq!(full.len(), 5);

        // Cursor by the last id of the first page; expect the next two.
        let first = stream_ids_for_uri(app.clone(), "/streams?limit=2").await;
        let cursor = first.last().expect("first page has a last id").clone();
        let expected_next: Vec<String> = full.iter().skip(2).take(2).cloned().collect();

        let uri = format!("/streams?after={cursor}&limit=2");
        let page = stream_ids_for_uri(app, &uri).await;
        assert_eq!(page.len(), 2, "limit caps the cursored page to two streams");
        assert_eq!(page, expected_next, "after-cursor selects the next slice");
        // Negative fixtures: items in the prior page must not reappear.
        assert!(
            !page.contains(&full[0]),
            "first-page item must not reappear"
        );
        assert!(
            !page.contains(&full[1]),
            "first-page item must not reappear"
        );
    }
}