klieo-ops-api 2.2.0

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

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

pub(crate) async fn list_streams(
    State(state): State<Arc<OpsState>>,
) -> Result<Json<Vec<StreamSummary>>, OpsError> {
    let buffer = state
        .resume_buffer
        .as_ref()
        .ok_or(OpsError::StoreNotConfigured)?;

    let entries = buffer.list_all_streams().await.map_err(|e| {
        tracing::error!(operation = "list_streams", detail = %e, "buffer error");
        OpsError::Internal(Box::new(e))
    })?;

    let summaries = 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 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::resume::{KvResumeBuffer, ResumeBuffer};
    use std::sync::Arc;
    use tower::ServiceExt;

    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_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);
    }
}