use crate::providers::response_stream::OpenedStream;
use crate::{LlmError, ProviderError, Result};
use eventsource_stream::Eventsource;
use futures::{Stream, StreamExt};
use reqwest::header::{ACCEPT, HeaderMap};
use reqwest::{Client, RequestBuilder, Response};
use serde::Serialize;
use serde_json::{Value, from_str};
use std::future::{Future, ready};
use std::pin::Pin;
use std::time::Duration;
pub(crate) type SseData = Pin<Box<dyn Stream<Item = Result<String>> + Send>>;
pub(crate) fn http_client() -> reqwest::Client {
reqwest::Client::builder().connect_timeout(Duration::from_secs(30)).build().expect("HTTP client should build")
}
pub(crate) fn post_sse<T: Serialize>(
client: &Client,
url: &str,
headers: HeaderMap,
body: &T,
code: fn(&str) -> Option<String>,
) -> impl Future<Output = Result<OpenedStream<SseData>>> + Send + use<T> {
open_sse(client.post(url).headers(headers).header(ACCEPT, "text/event-stream").json(body), code)
}
async fn open_sse(request: RequestBuilder, code: fn(&str) -> Option<String>) -> Result<OpenedStream<SseData>> {
let response = request.send().await?;
if !response.status().is_success() {
return Err(rejected(response, code).await.into());
}
let http = HttpResponseMetadata::from(&response);
let data: SseData = Box::pin(
response
.bytes_stream()
.eventsource()
.take_while(|event| ready(!matches!(event, Ok(event) if event.data == "[DONE]")))
.filter_map(|event| {
ready(match event {
Ok(event) if event.event == "keepalive" => None,
Ok(event) => Some(Ok(event.data)),
Err(error) => Some(Err(ProviderError::stream_interrupted(error.to_string()).into())),
})
}),
);
Ok(OpenedStream::new(data).with_http(http))
}
#[derive(Debug, Clone)]
pub(crate) struct HttpResponseMetadata {
pub(crate) status: u16,
pub(crate) request_id: Option<String>,
}
impl HttpResponseMetadata {
pub(crate) fn annotate(&self, error: LlmError) -> LlmError {
match error {
LlmError::Provider(provider) => {
provider.with_http_metadata(Some(self.status), self.request_id.clone()).into()
}
error => error,
}
}
}
impl From<&Response> for HttpResponseMetadata {
fn from(response: &Response) -> Self {
Self { status: response.status().as_u16(), request_id: extract_request_id(response.headers()) }
}
}
pub(crate) fn extract_request_id(headers: &HeaderMap) -> Option<String> {
for name in ["x-amzn-requestid", "x-amz-request-id", "x-request-id", "request-id"] {
if let Some(value) = headers.get(name)
&& let Ok(text) = value.to_str()
{
let trimmed = text.trim();
if !trimmed.is_empty() {
return Some(trimmed.to_string());
}
}
}
None
}
pub(crate) async fn rejected(response: Response, code: fn(&str) -> Option<String>) -> ProviderError {
let metadata = HttpResponseMetadata::from(&response);
let body = response.text().await.unwrap_or_else(|_| "Unknown error".to_string());
let message = format!("request failed with status {}: {body}", metadata.status);
ProviderError::from_http_status(metadata.status, message)
.with_code(code(&body))
.with_request_id(metadata.request_id)
}
pub(crate) fn extract_json_code(body: &str, pointer: &str) -> Option<String> {
match from_str::<Value>(body).ok()?.pointer(pointer)? {
Value::String(code) => Some(code.clone()),
Value::Number(code) => Some(code.to_string()),
_ => None,
}
}
pub(crate) fn responses_code(body: &str) -> Option<String> {
extract_json_code(body, "/error/code")
}
pub(crate) fn anthropic_code(body: &str) -> Option<String> {
extract_json_code(body, "/error/type")
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn request_id_prefers_amazon_headers() {
let mut headers = reqwest::header::HeaderMap::new();
headers.insert("x-request-id", "openai-1".parse().unwrap());
headers.insert("x-amzn-requestid", "amzn-1".parse().unwrap());
assert_eq!(extract_request_id(&headers).as_deref(), Some("amzn-1"));
}
#[test]
fn request_id_ignores_blank_values() {
let mut headers = reqwest::header::HeaderMap::new();
headers.insert("x-request-id", " ".parse().unwrap());
assert_eq!(extract_request_id(&headers), None);
}
#[test]
fn extracts_nested_json_codes() {
assert_eq!(anthropic_code(r#"{"error":{"type":"rate_limit_error"}}"#).as_deref(), Some("rate_limit_error"));
assert_eq!(responses_code(r#"{"error":{"code":"server_error"}}"#).as_deref(), Some("server_error"));
}
}