1use super::{FileCrudLimits, FileFailure, FileOperationError};
2use crate::{
3 HttpBody, WireRequest, WireResponse,
4 adapt::{JsonInvocation, invoke_empty},
5 capability::Upstream,
6 codec::{self, CodecLimits},
7 transform::{TransformError, TransformErrorKind},
8 wire::{DeclaredFields, claude::files as c, gemini::files as g, openai::files as o},
9};
10use std::collections::BTreeSet;
11
12fn template(request: &WireRequest<()>, method: http::Method) -> Result<(), TransformError> {
13 super::path(&request.path)?;
14 if request.method != method || request.query.as_ref().is_some_and(|v| !v.is_empty()) {
15 return Err(TransformError::shape(
16 "files.template",
17 "selected template method must match operation; pass filters using typed query",
18 ));
19 }
20 Ok(())
21}
22
23fn get_request(
24 mut request: WireRequest<()>,
25 id: &str,
26 gemini: bool,
27) -> Result<WireRequest<()>, TransformError> {
28 template(&request, http::Method::GET)?;
29 request.path = format!(
30 "{}/{}",
31 request.path.trim_end_matches('/'),
32 if gemini {
33 super::name(id, true)?
34 .strip_prefix("files/")
35 .unwrap()
36 .to_owned()
37 } else {
38 super::name(id, false)?
39 }
40 );
41 Ok(request)
42}
43
44pub async fn openai_get<U: Upstream>(
45 u: &U,
46 target: &U::Target,
47 template: WireRequest<()>,
48 id: &str,
49 limits: CodecLimits,
50) -> Result<JsonInvocation<o::FileObject>, TransformError> {
51 invoke_empty(u, target, get_request(template, id, false)?, limits).await
52}
53
54pub async fn claude_get<U: Upstream>(
55 u: &U,
56 target: &U::Target,
57 template: WireRequest<()>,
58 id: &str,
59 limits: CodecLimits,
60) -> Result<JsonInvocation<c::FileMetadata>, TransformError> {
61 invoke_empty(u, target, get_request(template, id, false)?, limits).await
62}
63
64pub async fn gemini_get<U: Upstream>(
65 u: &U,
66 target: &U::Target,
67 template: WireRequest<()>,
68 name: &str,
69 limits: CodecLimits,
70) -> Result<JsonInvocation<g::File>, TransformError> {
71 invoke_empty(u, target, get_request(template, name, true)?, limits).await
72}
73
74pub async fn delete_empty<U: Upstream>(
75 u: &U,
76 target: &U::Target,
77 mut request: WireRequest<()>,
78) -> Result<WireResponse<HttpBody>, TransformError> {
79 super::path(&request.path)?;
80 if request.method != http::Method::DELETE {
81 return Err(TransformError::shape(
82 "files.delete",
83 "DELETE template required",
84 ));
85 }
86 for name in [
87 http::header::CONTENT_LENGTH,
88 http::header::CONTENT_ENCODING,
89 http::header::TRANSFER_ENCODING,
90 http::header::CONTENT_TYPE,
91 ] {
92 request.headers.remove(name);
93 }
94 u.send(
95 target,
96 WireRequest {
97 method: request.method,
98 path: request.path,
99 query: request.query,
100 headers: request.headers,
101 body: HttpBody::Bytes(bytes::Bytes::new()),
102 },
103 )
104 .await
105 .map_err(TransformError::from)
106}
107
108struct Progress {
109 calls: usize,
110 files: usize,
111 bytes: u64,
112 ids: BTreeSet<String>,
113 tokens: BTreeSet<String>,
114 limits: FileCrudLimits,
115}
116
117impl Progress {
118 fn new(request: &WireRequest<()>, limits: FileCrudLimits) -> Result<Self, FileOperationError> {
119 template(request, http::Method::GET)?;
120
121 Ok(Self {
122 calls: 0,
123 files: 0,
124 bytes: 0,
125 ids: BTreeSet::new(),
126 tokens: BTreeSet::new(),
127 limits,
128 })
129 }
130 fn err(&self, error: TransformError) -> FileOperationError {
131 FileOperationError {
132 completed_calls: self.calls,
133 completed_files: self.files,
134 failure: FileFailure::Transform(error),
135 }
136 }
137 fn limit(&self, field: &str) -> FileOperationError {
138 self.err(TransformError::new(
139 TransformErrorKind::Limit,
140 field,
141 "file collection limit exceeded",
142 ))
143 }
144 async fn page<
145 U: Upstream,
146 T: serde::de::DeserializeOwned + DeclaredFields + serde::Serialize,
147 >(
148 &mut self,
149 u: &U,
150 target: &U::Target,
151 request: WireRequest<()>,
152 ) -> Result<T, FileOperationError> {
153 let response = invoke_empty(u, target, request, self.limits.codec)
154 .await
155 .map_err(|e| self.err(e))?;
156 let body = match response {
157 JsonInvocation::Success(response) => response.body,
158 JsonInvocation::Rejected(response) => {
159 return Err(FileOperationError {
160 completed_calls: self.calls,
161 completed_files: self.files,
162 failure: FileFailure::Rejected(Box::new(response)),
163 });
164 }
165 };
166 self.calls += 1;
167 let bytes = codec::encode_json(&body, self.limits.codec)
168 .map_err(|e| {
169 self.err(TransformError::new(
170 TransformErrorKind::Limit,
171 "files.bytes",
172 e.to_string(),
173 ))
174 })?
175 .len() as u64;
176 self.bytes = self
177 .bytes
178 .checked_add(bytes)
179 .ok_or_else(|| self.limit("files.bytes"))?;
180 if self.bytes > self.limits.max_declared_bytes {
181 return Err(self.limit("files.bytes"));
182 }
183 Ok(body)
184 }
185 fn item(&mut self, id: &str) -> Result<(), FileOperationError> {
186 if id.is_empty() || !self.ids.insert(id.to_owned()) {
187 return Err(self.err(TransformError::invalid_result(
188 "files.id",
189 "empty or duplicate file identity",
190 )));
191 }
192
193 self.files += 1;
194 Ok(())
195 }
196 fn cursor(&mut self, token: String) -> Result<String, FileOperationError> {
197 if token.is_empty() || !self.tokens.insert(token.clone()) {
198 return Err(self.err(TransformError::invalid_result(
199 "files.cursor",
200 "empty or repeated cursor",
201 )));
202 }
203 Ok(token)
204 }
205}
206
207fn page_request(request: &WireRequest<()>, pairs: Vec<(&str, String)>) -> WireRequest<()> {
208 WireRequest {
209 method: request.method.clone(),
210 path: request.path.clone(),
211 headers: request.headers.clone(),
212 query: (!pairs.is_empty()).then(|| {
213 pairs
214 .into_iter()
215 .map(|(k, v)| format!("{k}={}", super::component(&v)))
216 .collect::<Vec<_>>()
217 .join("&")
218 }),
219 body: (),
220 }
221}
222
223pub async fn openai_list<U: Upstream>(
224 u: &U,
225 target: &U::Target,
226 request: WireRequest<()>,
227 query: o::ListFilesQuery,
228 limits: FileCrudLimits,
229) -> Result<Vec<o::FileObject>, FileOperationError> {
230 let query = query.into_declared();
231 if query.limit.is_some_and(|v| v <= 0 || v > 10000) {
232 return Err(TransformError::shape("limit", "OpenAI list limit must be 1..10000").into());
233 }
234 let mut progress = Progress::new(&request, limits)?;
235 let mut cursor = query.after;
236 let mut out = Vec::new();
237 if let Some(cursor) = &cursor {
238 progress.tokens.insert(cursor.clone());
239 }
240 loop {
241 let mut pairs = Vec::new();
242 if let Some(value) = &cursor {
243 pairs.push(("after", value.clone()));
244 }
245 if let Some(value) = &query.purpose {
246 pairs.push(("purpose", value.clone()));
247 }
248 if let Some(value) = query.limit {
249 pairs.push(("limit", value.to_string()));
250 }
251 if let Some(value) = query.order {
252 pairs.push((
253 "order",
254 match value {
255 o::FileOrder::Asc => "asc",
256 o::FileOrder::Desc => "desc",
257 }
258 .into(),
259 ));
260 }
261 let page: o::ListFilesResponseBody = progress
262 .page(u, target, page_request(&request, pairs))
263 .await?;
264 if let (Some(first), Some(last)) = (page.data.first(), page.data.last()) {
265 if first.id != page.first_id || last.id != page.last_id {
266 return Err(progress.err(TransformError::invalid_result(
267 "pagination",
268 "file boundaries disagree",
269 )));
270 }
271 } else if page.has_more {
272 return Err(progress.err(TransformError::invalid_result(
273 "pagination",
274 "empty continuing page",
275 )));
276 }
277 for file in page.data {
278 progress.item(&file.id)?;
279 out.push(file);
280 }
281 if !page.has_more {
282 return Ok(out);
283 }
284 cursor = Some(progress.cursor(page.last_id)?);
285 }
286}
287
288pub async fn claude_list<U: Upstream>(
289 u: &U,
290 target: &U::Target,
291 request: WireRequest<()>,
292 query: c::ListFilesQuery,
293 limits: FileCrudLimits,
294) -> Result<Vec<c::FileMetadata>, FileOperationError> {
295 let query = query.into_declared();
296 if query.after_id.is_some() && query.before_id.is_some() {
297 return Err(TransformError::shape("cursor", "after_id and before_id conflict").into());
298 }
299 if query.limit.is_some_and(|v| v <= 0 || v > 1000) {
300 return Err(TransformError::shape("limit", "Claude limit must be 1..1000").into());
301 }
302 let backward = query.before_id.is_some();
303 let mut cursor = query.before_id.or(query.after_id);
304 let mut progress = Progress::new(&request, limits)?;
305 let mut out = Vec::new();
306 if let Some(cursor) = &cursor {
307 progress.tokens.insert(cursor.clone());
308 }
309 loop {
310 let mut pairs = Vec::new();
311 if let Some(cursor) = &cursor {
312 pairs.push((
313 if backward { "before_id" } else { "after_id" },
314 cursor.clone(),
315 ));
316 }
317 if let Some(scope) = &query.scope_id {
318 pairs.push(("scope_id", scope.clone()));
319 }
320 if let Some(limit) = query.limit {
321 pairs.push(("limit", limit.to_string()));
322 }
323 let page: c::ListFilesResponseBody = progress
324 .page(u, target, page_request(&request, pairs))
325 .await?;
326 let more = page
327 .has_more
328 .ok_or_else(|| progress.err(TransformError::missing_metadata("has_more")))?;
329 if let (Some(first), Some(last)) = (page.data.first(), page.data.last()) {
330 if page.first_id.as_deref().is_some_and(|v| v != first.id)
331 || page.last_id.as_deref().is_some_and(|v| v != last.id)
332 {
333 return Err(progress.err(TransformError::invalid_result(
334 "pagination",
335 "file boundaries disagree",
336 )));
337 }
338 } else if more {
339 return Err(progress.err(TransformError::invalid_result(
340 "pagination",
341 "empty continuing page",
342 )));
343 }
344 for file in page.data {
345 progress.item(&file.id)?;
346 out.push(file);
347 }
348 if !more {
349 return Ok(out);
350 }
351 cursor = Some(
352 progress.cursor(
353 if backward {
354 page.first_id
355 } else {
356 page.last_id
357 }
358 .ok_or_else(|| {
359 progress.err(TransformError::missing_metadata("pagination cursor"))
360 })?,
361 )?,
362 );
363 }
364}
365
366pub async fn gemini_list<U: Upstream>(
367 u: &U,
368 target: &U::Target,
369 request: WireRequest<()>,
370 query: g::ListFilesQuery,
371 limits: FileCrudLimits,
372) -> Result<Vec<g::File>, FileOperationError> {
373 let query = query.into_declared();
374 if query.page_size.flatten().is_some_and(|v| v <= 0 || v > 100) {
375 return Err(TransformError::shape("page_size", "Gemini file limit must be 1..100").into());
376 }
377 let mut cursor = query.page_token.flatten();
378 let mut progress = Progress::new(&request, limits)?;
379 let mut out = Vec::new();
380 if let Some(cursor) = &cursor {
381 progress.tokens.insert(cursor.clone());
382 }
383 loop {
384 let mut pairs = Vec::new();
385 if let Some(cursor) = &cursor {
386 pairs.push(("pageToken", cursor.clone()));
387 }
388 if let Some(size) = query.page_size.flatten() {
389 pairs.push(("pageSize", size.to_string()));
390 }
391 let page: g::ListFilesResponseBody = progress
392 .page(u, target, page_request(&request, pairs))
393 .await?;
394 for file in page.files.flatten().unwrap_or_default() {
395 let id = file
396 .name
397 .as_ref()
398 .and_then(Option::as_ref)
399 .ok_or_else(|| progress.err(TransformError::missing_metadata("file.name")))?;
400 super::name(id, true).map_err(|e| progress.err(e))?;
401 progress.item(id)?;
402 out.push(file);
403 }
404 match page.next_page_token.flatten().filter(|v| !v.is_empty()) {
405 None => return Ok(out),
406 Some(value) => cursor = Some(progress.cursor(value)?),
407 }
408 }
409}