chore: publish clean community baseline

crates/crank-adapter-rest/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "crank-adapter-rest"
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+async-trait = "0.1"
+crank-core = { path = "../crank-core" }
+futures-util = "0.3"
+reqwest = { workspace = true, features = ["stream"] }
+serde.workspace = true
+serde_json.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+
+[dev-dependencies]
+axum.workspace = true
+tokio.workspace = true

crates/crank-adapter-rest/src/client.rs

@@ -0,0 +1,463 @@
+use std::{collections::BTreeMap, time::Duration};
+
+use crank_core::{HttpMethod, RestTarget};
+use reqwest::{
+    Client,
+    header::{HeaderMap, HeaderName, HeaderValue},
+};
+use serde_json::Value;
+
+use crate::{RestAdapterError, RestRequest, RestResponse, RestWindowRequest, RestWindowResponse};
+
+#[derive(Clone, Debug)]
+pub struct RestAdapter {
+    client: Client,
+}
+
+impl Default for RestAdapter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl RestAdapter {
+    pub fn new() -> Self {
+        Self {
+            client: Client::new(),
+        }
+    }
+
+    pub async fn execute(
+        &self,
+        target: &RestTarget,
+        request: &RestRequest,
+    ) -> Result<RestResponse, RestAdapterError> {
+        let url = build_url(target, request)?;
+        let headers = build_headers(target, request)?;
+        let mut builder = self
+            .client
+            .request(to_reqwest_method(target.method), url)
+            .headers(headers)
+            .timeout(Duration::from_millis(request.timeout_ms));
+
+        if let Some(body) = &request.body {
+            builder = builder.json(body);
+        }
+
+        let response = builder.send().await?;
+        let status = response.status();
+        let headers = normalize_headers(response.headers());
+        let body = decode_body(response).await?;
+
+        if !status.is_success() {
+            return Err(RestAdapterError::UnexpectedStatus {
+                status: status.as_u16(),
+                body,
+            });
+        }
+
+        Ok(RestResponse {
+            status_code: status.as_u16(),
+            headers,
+            body,
+        })
+    }
+
+    pub async fn execute_window(
+        &self,
+        target: &RestTarget,
+        request: &RestWindowRequest,
+    ) -> Result<RestWindowResponse, RestAdapterError> {
+        let url = build_url(target, &request.request)?;
+        let mut headers = build_headers(target, &request.request)?;
+        headers.insert(
+            reqwest::header::ACCEPT,
+            HeaderValue::from_static("text/event-stream"),
+        );
+
+        let mut builder = self
+            .client
+            .request(to_reqwest_method(target.method), url)
+            .headers(headers)
+            .timeout(Duration::from_millis(request.request.timeout_ms));
+
+        if let Some(body) = &request.request.body {
+            builder = builder.json(body);
+        }
+
+        let response = builder.send().await?;
+        let status = response.status();
+
+        if !status.is_success() {
+            let headers = normalize_headers(response.headers());
+            let body = decode_body(response).await?;
+            return Err(RestAdapterError::UnexpectedStatus {
+                status: status.as_u16(),
+                body: Value::Object(
+                    [
+                        (
+                            "headers".to_owned(),
+                            serde_json::to_value(headers).unwrap_or(Value::Null),
+                        ),
+                        ("body".to_owned(), body),
+                    ]
+                    .into_iter()
+                    .collect(),
+                ),
+            });
+        }
+
+        let (status_code, headers, body) =
+            crate::sse::collect_sse_window(response, request.window_duration_ms, request.max_items)
+                .await?;
+
+        Ok(RestWindowResponse {
+            status_code,
+            headers,
+            body,
+        })
+    }
+}
+
+fn build_url(target: &RestTarget, request: &RestRequest) -> Result<reqwest::Url, RestAdapterError> {
+    let base_url =
+        reqwest::Url::parse(&target.base_url).map_err(|_| RestAdapterError::InvalidBaseUrl {
+            url: target.base_url.clone(),
+        })?;
+    let path = substitute_path_params(&target.path_template, &request.path_params);
+    let mut url = base_url.join(path.trim_start_matches('/')).map_err(|_| {
+        RestAdapterError::InvalidBaseUrl {
+            url: target.base_url.clone(),
+        }
+    })?;
+
+    {
+        let mut query = url.query_pairs_mut();
+        for (key, value) in &request.query_params {
+            query.append_pair(key, value);
+        }
+    }
+
+    Ok(url)
+}
+
+fn substitute_path_params(path_template: &str, path_params: &BTreeMap<String, String>) -> String {
+    let mut rendered = path_template.to_owned();
+
+    for (key, value) in path_params {
+        rendered = rendered.replace(&format!("{{{key}}}"), value);
+    }
+
+    rendered
+}
+
+fn build_headers(
+    target: &RestTarget,
+    request: &RestRequest,
+) -> Result<HeaderMap, RestAdapterError> {
+    let mut headers = HeaderMap::new();
+
+    for (name, value) in &target.static_headers {
+        insert_header(&mut headers, name, value)?;
+    }
+
+    for (name, value) in &request.headers {
+        insert_header(&mut headers, name, value)?;
+    }
+
+    Ok(headers)
+}
+
+fn insert_header(headers: &mut HeaderMap, name: &str, value: &str) -> Result<(), RestAdapterError> {
+    let header_name =
+        HeaderName::try_from(name).map_err(|_| RestAdapterError::InvalidHeaderName {
+            header: name.to_owned(),
+        })?;
+    let header_value =
+        HeaderValue::try_from(value).map_err(|_| RestAdapterError::InvalidHeaderValue {
+            header: name.to_owned(),
+        })?;
+
+    headers.insert(header_name, header_value);
+    Ok(())
+}
+
+async fn decode_body(response: reqwest::Response) -> Result<Value, RestAdapterError> {
+    let bytes = response.bytes().await?;
+
+    if bytes.is_empty() {
+        return Ok(Value::Null);
+    }
+
+    match serde_json::from_slice::<Value>(&bytes) {
+        Ok(value) => Ok(value),
+        Err(_) => Ok(Value::String(
+            String::from_utf8_lossy(&bytes).trim().to_owned(),
+        )),
+    }
+}
+
+fn normalize_headers(headers: &HeaderMap) -> BTreeMap<String, String> {
+    headers
+        .iter()
+        .filter_map(|(name, value)| {
+            value
+                .to_str()
+                .ok()
+                .map(|value| (name.as_str().to_owned(), value.to_owned()))
+        })
+        .collect()
+}
+
+fn to_reqwest_method(method: HttpMethod) -> reqwest::Method {
+    match method {
+        HttpMethod::Get => reqwest::Method::GET,
+        HttpMethod::Post => reqwest::Method::POST,
+        HttpMethod::Put => reqwest::Method::PUT,
+        HttpMethod::Patch => reqwest::Method::PATCH,
+        HttpMethod::Delete => reqwest::Method::DELETE,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use axum::{
+        Json, Router,
+        extract::{Path, Query},
+        http::HeaderMap,
+        response::sse::{Event, KeepAlive, Sse},
+        routing::{get, post},
+    };
+    use crank_core::{HttpMethod, RestTarget};
+    use futures_util::stream;
+    use serde_json::{Value, json};
+    use tokio::net::TcpListener;
+
+    use crate::{RestAdapter, RestAdapterError, RestRequest, RestWindowRequest};
+
+    #[tokio::test]
+    async fn executes_rest_request_and_normalizes_json_response() {
+        let base_url = spawn_test_server().await;
+        let adapter = RestAdapter::new();
+        let target = RestTarget {
+            base_url,
+            method: HttpMethod::Post,
+            path_template: "/users/{user_id}".to_owned(),
+            static_headers: BTreeMap::from([("x-static".to_owned(), "static".to_owned())]),
+        };
+        let request = RestRequest {
+            path_params: BTreeMap::from([("user_id".to_owned(), "42".to_owned())]),
+            query_params: BTreeMap::from([("expand".to_owned(), "true".to_owned())]),
+            headers: BTreeMap::from([("x-trace-id".to_owned(), "trace-123".to_owned())]),
+            body: Some(json!({ "name": "Ada" })),
+            timeout_ms: 1_000,
+        };
+
+        let response = adapter.execute(&target, &request).await.unwrap();
+
+        assert_eq!(response.status_code, 200);
+        assert_eq!(
+            response.body,
+            json!({
+                "id": "42",
+                "query": "true",
+                "trace": "trace-123",
+                "static": "static",
+                "payload": { "name": "Ada" }
+            })
+        );
+    }
+
+    #[tokio::test]
+    async fn returns_unexpected_status_with_normalized_body() {
+        let base_url = spawn_test_server().await;
+        let adapter = RestAdapter::new();
+        let target = RestTarget {
+            base_url,
+            method: HttpMethod::Get,
+            path_template: "/fail".to_owned(),
+            static_headers: BTreeMap::new(),
+        };
+        let request = RestRequest {
+            path_params: BTreeMap::new(),
+            query_params: BTreeMap::new(),
+            headers: BTreeMap::new(),
+            body: None,
+            timeout_ms: 1_000,
+        };
+
+        let error = adapter.execute(&target, &request).await.unwrap_err();
+
+        assert!(matches!(
+            error,
+            RestAdapterError::UnexpectedStatus {
+                status: 502,
+                body: Value::Object(_)
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn collects_sse_events_with_window_bounds() {
+        let base_url = spawn_test_server().await;
+        let adapter = RestAdapter::new();
+        let target = RestTarget {
+            base_url,
+            method: HttpMethod::Get,
+            path_template: "/events".to_owned(),
+            static_headers: BTreeMap::new(),
+        };
+        let request = RestWindowRequest {
+            request: RestRequest {
+                path_params: BTreeMap::new(),
+                query_params: BTreeMap::new(),
+                headers: BTreeMap::new(),
+                body: None,
+                timeout_ms: 1_000,
+            },
+            window_duration_ms: 1_000,
+            max_items: Some(2),
+        };
+
+        let response = adapter.execute_window(&target, &request).await.unwrap();
+
+        assert_eq!(response.status_code, 200);
+        assert_eq!(
+            response.body,
+            json!({
+                "items": [
+                    { "message": "one" },
+                    { "message": "two" }
+                ],
+                "done": false
+            })
+        );
+    }
+
+    #[tokio::test]
+    async fn returns_timeout_window_when_no_events_arrive_before_deadline() {
+        let base_url = spawn_test_server().await;
+        let adapter = RestAdapter::new();
+        let target = RestTarget {
+            base_url,
+            method: HttpMethod::Get,
+            path_template: "/events-idle".to_owned(),
+            static_headers: BTreeMap::new(),
+        };
+        let request = RestWindowRequest {
+            request: RestRequest {
+                path_params: BTreeMap::new(),
+                query_params: BTreeMap::new(),
+                headers: BTreeMap::new(),
+                body: None,
+                timeout_ms: 1_000,
+            },
+            window_duration_ms: 50,
+            max_items: Some(10),
+        };
+
+        let response = adapter.execute_window(&target, &request).await.unwrap();
+
+        assert_eq!(response.body, json!({ "items": [], "done": true }));
+    }
+
+    #[tokio::test]
+    async fn rejects_malformed_sse_payloads() {
+        let base_url = spawn_test_server().await;
+        let adapter = RestAdapter::new();
+        let target = RestTarget {
+            base_url,
+            method: HttpMethod::Get,
+            path_template: "/events-broken".to_owned(),
+            static_headers: BTreeMap::new(),
+        };
+        let request = RestWindowRequest {
+            request: RestRequest {
+                path_params: BTreeMap::new(),
+                query_params: BTreeMap::new(),
+                headers: BTreeMap::new(),
+                body: None,
+                timeout_ms: 1_000,
+            },
+            window_duration_ms: 1_000,
+            max_items: Some(10),
+        };
+
+        let error = adapter.execute_window(&target, &request).await.unwrap_err();
+
+        assert!(matches!(error, RestAdapterError::InvalidSseEvent));
+    }
+
+    async fn spawn_test_server() -> String {
+        let app = Router::new()
+            .route("/users/{user_id}", post(create_user))
+            .route("/fail", get(fail))
+            .route("/events", get(sse_events))
+            .route("/events-idle", get(sse_idle))
+            .route("/events-broken", get(sse_broken));
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+
+        format!("http://{}", address)
+    }
+
+    async fn create_user(
+        Path(user_id): Path<String>,
+        Query(query): Query<BTreeMap<String, String>>,
+        headers: HeaderMap,
+        Json(payload): Json<Value>,
+    ) -> Json<Value> {
+        let trace = headers
+            .get("x-trace-id")
+            .and_then(|value| value.to_str().ok())
+            .unwrap_or_default();
+        let static_header = headers
+            .get("x-static")
+            .and_then(|value| value.to_str().ok())
+            .unwrap_or_default();
+
+        Json(json!({
+            "id": user_id,
+            "query": query.get("expand").cloned().unwrap_or_default(),
+            "trace": trace,
+            "static": static_header,
+            "payload": payload
+        }))
+    }
+
+    async fn fail() -> (axum::http::StatusCode, Json<Value>) {
+        (
+            axum::http::StatusCode::BAD_GATEWAY,
+            Json(json!({ "error": "upstream failed" })),
+        )
+    }
+
+    async fn sse_events()
+    -> Sse<impl futures_util::Stream<Item = Result<Event, std::convert::Infallible>>> {
+        let events = vec![
+            Ok(Event::default().data("{\"message\":\"one\"}")),
+            Ok(Event::default().data("{\"message\":\"two\"}")),
+            Ok(Event::default().data("{\"message\":\"three\"}")),
+        ];
+
+        Sse::new(stream::iter(events)).keep_alive(KeepAlive::default())
+    }
+
+    async fn sse_idle()
+    -> Sse<impl futures_util::Stream<Item = Result<Event, std::convert::Infallible>>> {
+        Sse::new(stream::pending()).keep_alive(KeepAlive::default())
+    }
+
+    async fn sse_broken()
+    -> Sse<impl futures_util::Stream<Item = Result<Event, std::convert::Infallible>>> {
+        let events = vec![Ok(Event::default().data("{broken-json}"))];
+
+        Sse::new(stream::iter(events)).keep_alive(KeepAlive::default())
+    }
+}

crates/crank-adapter-rest/src/error.rs

@@ -0,0 +1,24 @@
+use serde_json::Value;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum RestAdapterError {
+    #[error("invalid base url: {url}")]
+    InvalidBaseUrl { url: String },
+    #[error("invalid path parameter {key}")]
+    InvalidPathParameter { key: String },
+    #[error("invalid query parameter {key}")]
+    InvalidQueryParameter { key: String },
+    #[error("invalid header name {header}")]
+    InvalidHeaderName { header: String },
+    #[error("invalid header value for {header}")]
+    InvalidHeaderValue { header: String },
+    #[error("request failed")]
+    Transport(#[from] reqwest::Error),
+    #[error("sse collection window expired before stream completed")]
+    WindowExpired,
+    #[error("rest endpoint returned status {status}")]
+    UnexpectedStatus { status: u16, body: Value },
+    #[error("sse stream produced malformed event payload")]
+    InvalidSseEvent,
+}

crates/crank-adapter-rest/src/lib.rs

@@ -0,0 +1,123 @@
+mod client;
+mod error;
+mod model;
+mod sse;
+
+use async_trait::async_trait;
+use crank_core::{
+    AdapterResponse, ExecutionMode, PreparedRequest, Protocol, ProtocolAdapter,
+    ProtocolAdapterError, RestTarget, RuntimeRequestContext, Target, WindowExecutionResult,
+};
+use serde_json::{Map, Value};
+
+pub use client::RestAdapter;
+pub use error::RestAdapterError;
+pub use model::{RestRequest, RestResponse, RestWindowRequest, RestWindowResponse};
+
+#[async_trait]
+impl ProtocolAdapter for RestAdapter {
+    fn protocol(&self) -> Protocol {
+        Protocol::Rest
+    }
+
+    fn supports_mode(&self, mode: ExecutionMode) -> bool {
+        matches!(mode, ExecutionMode::Unary | ExecutionMode::Window)
+    }
+
+    async fn invoke_unary(
+        &self,
+        target: &Target,
+        prepared: &PreparedRequest,
+        _context: &RuntimeRequestContext,
+    ) -> Result<AdapterResponse, ProtocolAdapterError> {
+        let target = rest_target(target)?;
+        let request = RestRequest {
+            path_params: prepared.path_params.clone(),
+            query_params: prepared.query_params.clone(),
+            headers: prepared.headers.clone(),
+            body: prepared.body.clone(),
+            timeout_ms: prepared.timeout_ms,
+        };
+        let response = self.execute(target, &request).await?;
+
+        Ok(AdapterResponse {
+            status_code: response.status_code,
+            headers: response.headers,
+            data: response.body.clone(),
+            body: response.body,
+        })
+    }
+
+    async fn invoke_window(
+        &self,
+        target: &Target,
+        prepared: &PreparedRequest,
+        window_duration_ms: u64,
+        max_items: Option<u32>,
+        _context: &RuntimeRequestContext,
+    ) -> Result<WindowExecutionResult, ProtocolAdapterError> {
+        let target = rest_target(target)?;
+        let request = RestWindowRequest {
+            request: RestRequest {
+                path_params: prepared.path_params.clone(),
+                query_params: prepared.query_params.clone(),
+                headers: prepared.headers.clone(),
+                body: prepared.body.clone(),
+                timeout_ms: prepared.timeout_ms,
+            },
+            window_duration_ms,
+            max_items: max_items.map(|value| value.saturating_add(1)),
+        };
+        let response = self.execute_window(target, &request).await?;
+        Ok(window_result_from_response(response.body, max_items))
+    }
+}
+
+fn rest_target(target: &Target) -> Result<&RestTarget, ProtocolAdapterError> {
+    match target {
+        Target::Rest(target) => Ok(target),
+        other => Err(ProtocolAdapterError::Message(format!(
+            "rest adapter cannot handle target {other:?}"
+        ))),
+    }
+}
+
+fn window_result_from_response(body: Value, max_items: Option<u32>) -> WindowExecutionResult {
+    let done = body.get("done").and_then(Value::as_bool).unwrap_or(true);
+    let mut items = body
+        .get("items")
+        .and_then(Value::as_array)
+        .cloned()
+        .unwrap_or_default();
+    let mut truncated = false;
+    let mut has_more = !done;
+
+    if let Some(max_items) = max_items.map(|value| value as usize) {
+        if items.len() > max_items {
+            items.truncate(max_items);
+            truncated = true;
+            has_more = true;
+        }
+    }
+
+    let summary = body
+        .get("summary")
+        .cloned()
+        .unwrap_or_else(|| Value::Object(Map::new()));
+    let cursor = body.get("cursor").cloned();
+
+    WindowExecutionResult {
+        summary,
+        items,
+        cursor,
+        window_complete: !has_more,
+        truncated,
+        has_more,
+    }
+}
+
+impl From<RestAdapterError> for ProtocolAdapterError {
+    fn from(value: RestAdapterError) -> Self {
+        ProtocolAdapterError::Message(value.to_string())
+    }
+}

crates/crank-adapter-rest/src/model.rs

@@ -0,0 +1,42 @@
+use std::collections::BTreeMap;
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
+pub struct RestRequest {
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub path_params: BTreeMap<String, String>,
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub query_params: BTreeMap<String, String>,
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub headers: BTreeMap<String, String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub body: Option<Value>,
+    pub timeout_ms: u64,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct RestResponse {
+    pub status_code: u16,
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub headers: BTreeMap<String, String>,
+    pub body: Value,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
+pub struct RestWindowRequest {
+    #[serde(flatten)]
+    pub request: RestRequest,
+    pub window_duration_ms: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub max_items: Option<u32>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct RestWindowResponse {
+    pub status_code: u16,
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub headers: BTreeMap<String, String>,
+    pub body: Value,
+}

crates/crank-adapter-rest/src/sse.rs

@@ -0,0 +1,160 @@
+use std::collections::BTreeMap;
+
+use futures_util::StreamExt;
+use reqwest::header::HeaderMap;
+use serde_json::{Value, json};
+use tokio::time::{Duration, Instant, timeout_at};
+
+use crate::RestAdapterError;
+
+pub async fn collect_sse_window(
+    response: reqwest::Response,
+    window_duration_ms: u64,
+    max_items: Option<u32>,
+) -> Result<(u16, BTreeMap<String, String>, Value), RestAdapterError> {
+    let status = response.status();
+    let headers = normalize_headers(response.headers());
+    let deadline = Instant::now() + Duration::from_millis(window_duration_ms);
+    let mut stream = response.bytes_stream();
+    let mut buffer = String::new();
+    let mut items = Vec::new();
+    let mut done = true;
+
+    loop {
+        if max_items.is_some_and(|limit| items.len() >= limit as usize) {
+            done = false;
+            break;
+        }
+
+        let next_chunk = match timeout_at(deadline, stream.next()).await {
+            Ok(next_chunk) => next_chunk,
+            Err(_) => break,
+        };
+
+        let Some(next_chunk) = next_chunk else {
+            break;
+        };
+
+        let chunk = next_chunk?;
+        buffer.push_str(&String::from_utf8_lossy(&chunk));
+
+        while let Some(event_end) = find_event_boundary(&buffer) {
+            let event = buffer[..event_end].to_owned();
+            let boundary_len = boundary_length(&buffer[event_end..]);
+            buffer = buffer[event_end + boundary_len..].to_owned();
+
+            if let Some(item) = parse_sse_event(&event)? {
+                items.push(item);
+
+                if max_items.is_some_and(|limit| items.len() >= limit as usize) {
+                    done = false;
+                    break;
+                }
+            }
+        }
+
+        if !done {
+            break;
+        }
+    }
+
+    Ok((
+        status.as_u16(),
+        headers,
+        json!({
+            "items": items,
+            "done": done
+        }),
+    ))
+}
+
+fn parse_sse_event(raw: &str) -> Result<Option<Value>, RestAdapterError> {
+    let mut data_lines = Vec::new();
+
+    for line in raw.lines() {
+        if line.is_empty() || line.starts_with(':') {
+            continue;
+        }
+
+        if let Some(value) = line.strip_prefix("data:") {
+            data_lines.push(value.trim_start().to_owned());
+        }
+    }
+
+    if data_lines.is_empty() {
+        return Ok(None);
+    }
+
+    let payload = data_lines.join("\n");
+    if payload.is_empty() {
+        return Ok(None);
+    }
+
+    serde_json::from_str::<Value>(&payload)
+        .map(Some)
+        .or_else(|_| {
+            if payload.starts_with('{') || payload.starts_with('[') {
+                Err(RestAdapterError::InvalidSseEvent)
+            } else {
+                Ok(Some(Value::String(payload)))
+            }
+        })
+}
+
+fn find_event_boundary(buffer: &str) -> Option<usize> {
+    buffer
+        .find("\r\n\r\n")
+        .or_else(|| buffer.find("\n\n"))
+        .or_else(|| buffer.find("\r\r"))
+}
+
+fn boundary_length(boundary: &str) -> usize {
+    if boundary.starts_with("\r\n\r\n") {
+        4
+    } else {
+        2
+    }
+}
+
+fn normalize_headers(headers: &HeaderMap) -> BTreeMap<String, String> {
+    headers
+        .iter()
+        .filter_map(|(name, value)| {
+            value
+                .to_str()
+                .ok()
+                .map(|value| (name.as_str().to_owned(), value.to_owned()))
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::json;
+
+    use super::parse_sse_event;
+    use crate::RestAdapterError;
+
+    #[test]
+    fn parses_json_event_payload() {
+        let event = "event: message\ndata: {\"message\":\"ok\"}\n\n";
+
+        let parsed = parse_sse_event(event).unwrap();
+
+        assert_eq!(parsed, Some(json!({ "message": "ok" })));
+    }
+
+    #[test]
+    fn ignores_comment_only_events() {
+        let parsed = parse_sse_event(": keepalive\n\n").unwrap();
+
+        assert_eq!(parsed, None);
+    }
+
+    #[test]
+    fn rejects_malformed_json_like_payload() {
+        let error = parse_sse_event("data: {broken-json}\n\n").unwrap_err();
+
+        assert!(matches!(error, RestAdapterError::InvalidSseEvent));
+    }
+}