sim_lib_openai_server/routes/
threads.rs

1use serde_json::{Value, json};
2
3use crate::{
4    clock::{GatewayClock, SystemGatewayClock},
5    ids::GatewayIdGenerator,
6    objects::{GatewayRequest, GatewayResponse},
7    server::GatewayRouteState,
8    storage::{GatewayStateStore, GatewayThread, GatewayThreadMessage},
9};
10
11use super::errors::OpenAiRouteError;
12
13/// Route path for thread creation (`POST /v1/threads`).
14pub const THREADS_PATH: &str = "/v1/threads";
15/// Path prefix shared by thread retrieval and message routes (`/v1/threads/`).
16pub const THREAD_RETRIEVAL_PREFIX: &str = "/v1/threads/";
17/// Templated route for retrieving a single thread by id (`/v1/threads/{id}`).
18pub const THREAD_RETRIEVAL_ROUTE: &str = "/v1/threads/{id}";
19/// Templated route for a thread's message collection (`/v1/threads/{id}/messages`).
20pub const THREAD_MESSAGES_ROUTE: &str = "/v1/threads/{id}/messages";
21
22type RouteResult<T> = std::result::Result<T, OpenAiRouteError>;
23
24/// Handles `POST /v1/threads`, creating a new thread and returning its JSON object.
25pub fn handle_threads(request: &GatewayRequest, state: &GatewayRouteState) -> GatewayResponse {
26    let mut clock = SystemGatewayClock;
27    let seed = clock.now_ms().unwrap_or(1);
28    let mut ids = GatewayIdGenerator::deterministic("thread", seed);
29    match state.store().lock() {
30        Ok(mut store) => create_thread(&mut *store, &mut ids, &mut clock, request)
31            .unwrap_or_else(OpenAiRouteError::into_response),
32        Err(err) => OpenAiRouteError::internal_message(format!("gateway store lock failed: {err}"))
33            .into_response(),
34    }
35}
36
37/// Handles `POST /v1/threads/{id}/messages`, appending a message to the thread.
38pub fn handle_thread_post(request: &GatewayRequest, state: &GatewayRouteState) -> GatewayResponse {
39    let Some(thread_id) = message_thread_id_from_path(request.path()) else {
40        return OpenAiRouteError::not_found_kind("thread", request.path()).into_response();
41    };
42    let mut clock = SystemGatewayClock;
43    let seed = clock.now_ms().unwrap_or(1);
44    let mut ids = GatewayIdGenerator::deterministic("msg", seed);
45    match state.store().lock() {
46        Ok(mut store) => append_message(&mut *store, &mut ids, &mut clock, thread_id, request)
47            .unwrap_or_else(OpenAiRouteError::into_response),
48        Err(err) => OpenAiRouteError::internal_message(format!("gateway store lock failed: {err}"))
49            .into_response(),
50    }
51}
52
53/// Handles `GET /v1/threads/{id}` and `GET /v1/threads/{id}/messages`, dispatching
54/// to thread retrieval or message listing based on the request path.
55pub fn handle_thread_get(request: &GatewayRequest, state: &GatewayRouteState) -> GatewayResponse {
56    match thread_path(request.path()) {
57        Some(ThreadPath::Thread(thread_id)) => match state.store().lock() {
58            Ok(store) => retrieve_thread(&*store, thread_id),
59            Err(err) => {
60                OpenAiRouteError::internal_message(format!("gateway store lock failed: {err}"))
61                    .into_response()
62            }
63        },
64        Some(ThreadPath::Messages(thread_id)) => match state.store().lock() {
65            Ok(store) => list_messages(&*store, thread_id),
66            Err(err) => {
67                OpenAiRouteError::internal_message(format!("gateway store lock failed: {err}"))
68                    .into_response()
69            }
70        },
71        None => OpenAiRouteError::not_found_kind("thread", request.path()).into_response(),
72    }
73}
74
75/// Returns the JSON object for a stored thread, or a not-found error response.
76pub fn retrieve_thread<S>(store: &S, thread_id: &str) -> GatewayResponse
77where
78    S: GatewayStateStore,
79{
80    store
81        .thread(thread_id)
82        .map(|thread| GatewayResponse::json(200, thread_json(&thread).to_string().into_bytes()))
83        .unwrap_or_else(|| OpenAiRouteError::not_found_kind("thread", thread_id).into_response())
84}
85
86/// Returns the thread's messages as an OpenAI `list` object, or a not-found error
87/// response when the thread does not exist.
88pub fn list_messages<S>(store: &S, thread_id: &str) -> GatewayResponse
89where
90    S: GatewayStateStore,
91{
92    if store.thread(thread_id).is_none() {
93        return OpenAiRouteError::not_found_kind("thread", thread_id).into_response();
94    }
95    let data = store
96        .thread_messages(thread_id)
97        .iter()
98        .map(message_json)
99        .collect::<Vec<_>>();
100    GatewayResponse::json(
101        200,
102        json!({ "object": "list", "data": data })
103            .to_string()
104            .into_bytes(),
105    )
106}
107
108fn create_thread<S, C>(
109    store: &mut S,
110    ids: &mut GatewayIdGenerator,
111    clock: &mut C,
112    request: &GatewayRequest,
113) -> RouteResult<GatewayResponse>
114where
115    S: GatewayStateStore,
116    C: GatewayClock,
117{
118    let object = request_object(request.body())?;
119    let thread = GatewayThread::new(
120        ids.next_id().map_err(OpenAiRouteError::internal)?,
121        clock.now_ms().map_err(OpenAiRouteError::internal)?,
122        metadata(object.get("metadata"))?,
123    );
124    store
125        .put_thread(thread.clone())
126        .map_err(OpenAiRouteError::internal)?;
127    Ok(GatewayResponse::json(
128        200,
129        thread_json(&thread).to_string().into_bytes(),
130    ))
131}
132
133fn append_message<S, C>(
134    store: &mut S,
135    ids: &mut GatewayIdGenerator,
136    clock: &mut C,
137    thread_id: &str,
138    request: &GatewayRequest,
139) -> RouteResult<GatewayResponse>
140where
141    S: GatewayStateStore,
142    C: GatewayClock,
143{
144    if store.thread(thread_id).is_none() {
145        return Err(OpenAiRouteError::not_found_kind("thread", thread_id));
146    }
147    let object = request_object(request.body())?;
148    let role = required_string(&object, "role")?.to_owned();
149    let content = required_string(&object, "content")?.to_owned();
150    let message = GatewayThreadMessage::new(
151        ids.next_id().map_err(OpenAiRouteError::internal)?,
152        thread_id,
153        role,
154        content,
155        clock.now_ms().map_err(OpenAiRouteError::internal)?,
156    );
157    store
158        .put_thread_message(message.clone())
159        .map_err(OpenAiRouteError::internal)?;
160    Ok(GatewayResponse::json(
161        200,
162        message_json(&message).to_string().into_bytes(),
163    ))
164}
165
166use crate::routes::request_json::{request_object_or_empty as request_object, required_string};
167
168fn metadata(value: Option<&Value>) -> RouteResult<Vec<(String, String)>> {
169    let Some(value) = value else {
170        return Ok(Vec::new());
171    };
172    let object = value.as_object().ok_or_else(|| {
173        OpenAiRouteError::bad_request(
174            "metadata must be an object",
175            Some("metadata"),
176            "invalid_metadata",
177        )
178    })?;
179    object
180        .iter()
181        .map(|(key, value)| {
182            value
183                .as_str()
184                .map(|value| (key.clone(), value.to_owned()))
185                .ok_or_else(|| {
186                    OpenAiRouteError::bad_request(
187                        "metadata values must be strings",
188                        Some("metadata"),
189                        "invalid_metadata",
190                    )
191                })
192        })
193        .collect()
194}
195
196enum ThreadPath<'a> {
197    Thread(&'a str),
198    Messages(&'a str),
199}
200
201fn thread_path(path: &str) -> Option<ThreadPath<'_>> {
202    if let Some(thread_id) = message_thread_id_from_path(path) {
203        return Some(ThreadPath::Messages(thread_id));
204    }
205    super::path::id_from_path(path, THREAD_RETRIEVAL_PREFIX).map(ThreadPath::Thread)
206}
207
208fn message_thread_id_from_path(path: &str) -> Option<&str> {
209    super::path::id_from_path_with_suffix(path, THREAD_RETRIEVAL_PREFIX, "/messages")
210}
211
212fn thread_json(thread: &GatewayThread) -> Value {
213    json!({
214        "id": thread.id(),
215        "object": "thread",
216        "created_at": thread.created_at_ms(),
217        "metadata": metadata_json(thread.metadata()),
218    })
219}
220
221fn message_json(message: &GatewayThreadMessage) -> Value {
222    json!({
223        "id": message.id(),
224        "object": "thread.message",
225        "thread_id": message.thread_id(),
226        "role": message.role(),
227        "content": message.content(),
228        "created_at": message.created_at_ms(),
229    })
230}
231
232fn metadata_json(metadata: &[(String, String)]) -> Value {
233    Value::Object(
234        metadata
235            .iter()
236            .map(|(key, value)| (key.clone(), Value::String(value.clone())))
237            .collect(),
238    )
239}
sim_lib_openai_server/routes/threads.rs

sim_lib_openai_server/routes/
threads.rs
