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}, Router, routing::get};
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);
}
}