gproxy_protocol/adapt/video/reverse/
invoke.rs

1use super::super::{composed::prepare_headers, direct::check_template, resources::bound_value};
2use super::*;
3
4fn expiry_valid(expiry: SystemTime) -> Result<(), TransformError> {
5    if expiry <= SystemTime::now() {
6        return Err(TransformError::shape(
7            "video.expiry",
8            "future publication expiry required",
9        ));
10    }
11    Ok(())
12}
13
14fn template(
15    binding: &ReverseVideoBinding,
16    mut template: WireRequest<()>,
17    method: http::Method,
18    id: Option<&str>,
19) -> Result<WireRequest<()>, TransformError> {
20    check_template(&template, method.clone())?;
21    let path = if let Some(id) = id {
22        binding.query_path(id)?
23    } else {
24        binding.create_path.clone()
25    };
26    if template.path != path {
27        return Err(TransformError::shape(
28            "video.template.path",
29            "path differs from saved target API binding",
30        ));
31    }
32    template.method = method;
33    Ok(template)
34}
35
36#[allow(clippy::too_many_arguments)]
37async fn send<U: Upstream>(
38    upstream: &U,
39    target: &U::Target,
40    template: WireRequest<()>,
41    request: Option<&ReverseVideoRequest>,
42    kind: ReverseVideoKind,
43    progress: &mut ReverseVideoProgress,
44    limits: VideoLimits,
45) -> Result<JsonInvocation<ReverseVideoResult>, TransformError> {
46    let mut codec = limits.codec;
47    codec.max_body_bytes = codec.max_body_bytes.min(upstream.limits().write_bytes);
48    codec.max_buffer_bytes = codec.max_buffer_bytes.min(codec.max_body_bytes);
49    codec.max_value_bytes = codec.max_value_bytes.min(codec.max_body_bytes);
50    let bytes = match request {
51        Some(ReverseVideoRequest::Native(v)) => crate::codec::encode_json(v, codec),
52        Some(ReverseVideoRequest::OpenRouter(v)) => crate::codec::encode_json(v, codec),
53        None => Ok(bytes::Bytes::new()),
54    }
55    .map_err(|e| TransformError::new(TransformErrorKind::Limit, "video.request", e.to_string()))?;
56    let request = prepare_headers(template, bytes, request.is_some());
57    progress.raw_response = None;
58    progress.result = None;
59    macro_rules! invoke {
60        ($ty:ty,$variant:ident) => {{
61            let mut native = None;
62            let response = super::super::transport::invoke::<_, $ty>(
63                upstream,
64                target,
65                request,
66                &mut progress.send_started,
67                &mut progress.raw_response,
68                &mut native,
69                limits,
70            )
71            .await;
72            progress.result = native.map(|r| ReverseVideoResult::$variant(r.body));
73            match response? {
74                JsonInvocation::Rejected(r) => Ok(JsonInvocation::Rejected(r)),
75                JsonInvocation::Success(r) => Ok(JsonInvocation::Success(WireResponse {
76                    status: r.status,
77                    headers: r.headers,
78                    body: ReverseVideoResult::$variant(r.body),
79                })),
80            }
81        }};
82    }
83    match kind {
84        ReverseVideoKind::Native => invoke!(o::NativeVideo, Native),
85        ReverseVideoKind::OpenRouter => invoke!(o::VideoGenerationResponseBody, OpenRouter),
86    }
87}
88
89#[allow(clippy::too_many_arguments)]
90pub async fn gemini_video_create_composed<U: Upstream, R: ResourceAccess, S: StateStore>(
91    upstream: &U,
92    target: &U::Target,
93    access: &R,
94    source_scope: &R::Scope,
95    target_scope: &R::Scope,
96    publish_scope: &R::Scope,
97    store: &S,
98    state_scope: &S::Scope,
99    create_template: WireRequest<()>,
100    input: g::PredictLongRunningRequestBody,
101    binding: ReverseVideoBinding,
102    effective_samples: Option<usize>,
103    defaults: Option<video::NativeVideoDefaults>,
104    expiry: SystemTime,
105    progress: &mut ReverseVideoProgress,
106    limits: VideoLimits,
107) -> Result<JsonInvocation<Converted<g::VideoOperation>>, TransformError> {
108    if progress.state.is_some() || progress.send_started {
109        return Err(conflict("video.progress"));
110    }
111    expiry_valid(expiry)?;
112    let state = prepare::split(
113        input,
114        binding,
115        effective_samples,
116        defaults.clone(),
117        expiry,
118        limits,
119    )?;
120    template(
121        &state.binding,
122        copy_template(&create_template),
123        http::Method::POST,
124        None,
125    )?;
126    prepare::controls(&state, defaults.clone())?;
127    save(store, state_scope, &state, progress, limits).await?;
128    drive(
129        upstream,
130        target,
131        access,
132        source_scope,
133        target_scope,
134        publish_scope,
135        store,
136        state_scope,
137        create_template,
138        state,
139        defaults,
140        expiry,
141        progress,
142        limits,
143    )
144    .await
145}
146
147/// Continue only children whose durable `started` marker is false. A started
148/// child without a retained result requires explicit caller reconciliation.
149#[allow(clippy::too_many_arguments)]
150pub async fn gemini_video_resume_composed<U: Upstream, R: ResourceAccess, S: StateStore>(
151    upstream: &U,
152    target: &U::Target,
153    access: &R,
154    source_scope: &R::Scope,
155    target_scope: &R::Scope,
156    publish_scope: &R::Scope,
157    store: &S,
158    state_scope: &S::Scope,
159    create_template: WireRequest<()>,
160    binding: ReverseVideoBinding,
161    defaults: Option<video::NativeVideoDefaults>,
162    expiry: SystemTime,
163    progress: &mut ReverseVideoProgress,
164    limits: VideoLimits,
165) -> Result<JsonInvocation<Converted<g::VideoOperation>>, TransformError> {
166    expiry_valid(expiry)?;
167    let (state, version) = load(store, state_scope, &binding, limits).await?;
168    if state.native_defaults != defaults || state.expires_at != expiry {
169        return Err(TransformError::shape(
170            "video.resume.facts",
171            "defaults and publication expiry differ from saved invocation",
172        ));
173    }
174    template(
175        &binding,
176        copy_template(&create_template),
177        http::Method::POST,
178        None,
179    )?;
180    prepare::controls(&state, defaults.clone())?;
181    progress.version = Some(version);
182    progress.state = Some(state.clone());
183    drive(
184        upstream,
185        target,
186        access,
187        source_scope,
188        target_scope,
189        publish_scope,
190        store,
191        state_scope,
192        create_template,
193        state,
194        defaults,
195        expiry,
196        progress,
197        limits,
198    )
199    .await
200}
201
202#[allow(clippy::too_many_arguments)]
203async fn drive<U: Upstream, R: ResourceAccess, S: StateStore>(
204    upstream: &U,
205    target: &U::Target,
206    access: &R,
207    source_scope: &R::Scope,
208    target_scope: &R::Scope,
209    publish_scope: &R::Scope,
210    store: &S,
211    state_scope: &S::Scope,
212    create_template: WireRequest<()>,
213    mut state: ReverseVideoState,
214    defaults: Option<video::NativeVideoDefaults>,
215    expiry: SystemTime,
216    progress: &mut ReverseVideoProgress,
217    limits: VideoLimits,
218) -> Result<JsonInvocation<Converted<g::VideoOperation>>, TransformError> {
219    let mut remaining = limits.codec.max_body_bytes;
220    for index in 0..state.children.len() {
221        if state.children[index].result.is_some() {
222            continue;
223        }
224        if state.children[index].started {
225            return Err(conflict("video.child.creation"));
226        }
227        progress.child_index = Some(index);
228        if state.children[index].request.is_none() {
229            state.children[index].request = Some(
230                prepare::prepare(
231                    access,
232                    source_scope,
233                    publish_scope,
234                    &state,
235                    index,
236                    defaults.clone(),
237                    expiry,
238                    progress,
239                    limits,
240                    &mut remaining,
241                )
242                .await?,
243            );
244            save(store, state_scope, &state, progress, limits).await?;
245        }
246        // Local encoding/write limits must fail before the durable send marker.
247        let mut preflight = limits.codec;
248        preflight.max_body_bytes = preflight.max_body_bytes.min(upstream.limits().write_bytes);
249        preflight.max_buffer_bytes = preflight.max_buffer_bytes.min(preflight.max_body_bytes);
250        preflight.max_value_bytes = preflight.max_value_bytes.min(preflight.max_body_bytes);
251        match state.children[index].request.as_ref().expect("prepared") {
252            ReverseVideoRequest::Native(v) => crate::codec::encode_json(v, preflight),
253            ReverseVideoRequest::OpenRouter(v) => crate::codec::encode_json(v, preflight),
254        }
255        .map_err(|e| {
256            TransformError::new(TransformErrorKind::Limit, "video.request", e.to_string())
257        })?;
258        state.children[index].started = true;
259        save(store, state_scope, &state, progress, limits).await?;
260        let response = send(
261            upstream,
262            target,
263            copy_template(&create_template),
264            state.children[index].request.as_ref(),
265            state.binding.kind,
266            progress,
267            limits,
268        )
269        .await?;
270        let response = match response {
271            JsonInvocation::Success(response) => response,
272            JsonInvocation::Rejected(r) => return Ok(JsonInvocation::Rejected(r)),
273        };
274
275        state.children[index].result = Some(response.body);
276        save(store, state_scope, &state, progress, limits).await?;
277    }
278    let body = output::finish(
279        access,
280        target_scope,
281        publish_scope,
282        store,
283        state_scope,
284        &mut state,
285        expiry,
286        progress,
287        limits,
288    )
289    .await?;
290    Ok(JsonInvocation::Success(WireResponse {
291        status: http::StatusCode::OK,
292        headers: http::HeaderMap::new(),
293        body,
294    }))
295}
296
297/// Poll a selected child using its exact recorded resource path. Other children
298/// retain their actual outcomes and are never implicitly recreated or polled.
299#[allow(clippy::too_many_arguments)]
300pub async fn gemini_video_query_composed<U: Upstream, R: ResourceAccess, S: StateStore>(
301    upstream: &U,
302    target: &U::Target,
303    access: &R,
304    target_scope: &R::Scope,
305    publish_scope: &R::Scope,
306    store: &S,
307    state_scope: &S::Scope,
308    query_template: WireRequest<()>,
309    binding: ReverseVideoBinding,
310    index: usize,
311    expiry: SystemTime,
312    progress: &mut ReverseVideoProgress,
313    limits: VideoLimits,
314) -> Result<JsonInvocation<Converted<g::VideoOperation>>, TransformError> {
315    expiry_valid(expiry)?;
316    let (mut state, version) = load(store, state_scope, &binding, limits).await?;
317    if state.expires_at != expiry {
318        return Err(TransformError::shape(
319            "video.query.expiry",
320            "publication expiry differs from saved invocation",
321        ));
322    }
323    let child = state
324        .children
325        .get(index)
326        .ok_or_else(|| TransformError::shape("video.child", "index out of range"))?;
327    let old = child
328        .result
329        .as_ref()
330        .ok_or_else(|| conflict("video.child.creation"))?;
331    let query_template = template(&binding, query_template, http::Method::GET, Some(old.id()))?;
332    progress.version = Some(version);
333    progress.state = Some(state.clone());
334    progress.child_index = Some(index);
335    let response = send(
336        upstream,
337        target,
338        query_template,
339        None,
340        binding.kind,
341        progress,
342        limits,
343    )
344    .await?;
345    let response = match response {
346        JsonInvocation::Success(response) => response,
347        JsonInvocation::Rejected(r) => return Ok(JsonInvocation::Rejected(r)),
348    };
349
350    if state.children[index]
351        .result
352        .as_ref()
353        .is_none_or(|old| !old.same_projection(&response.body))
354    {
355        state.children[index].published = None;
356    }
357    state.children[index].result = Some(response.body);
358    save(store, state_scope, &state, progress, limits).await?;
359    let body = output::finish(
360        access,
361        target_scope,
362        publish_scope,
363        store,
364        state_scope,
365        &mut state,
366        expiry,
367        progress,
368        limits,
369    )
370    .await?;
371    Ok(JsonInvocation::Success(WireResponse {
372        status: response.status,
373        headers: response.headers,
374        body,
375    }))
376}
377
378/// Persist a retained, real result after a post-send CAS failure. This performs
379/// no upstream request; unknown outcomes need caller reconciliation first.
380pub async fn recover_reverse_video_result<S: StateStore>(
381    store: &S,
382    scope: &S::Scope,
383    binding: &ReverseVideoBinding,
384    progress: &mut ReverseVideoProgress,
385    limits: VideoLimits,
386) -> Result<(), TransformError> {
387    let index = progress
388        .child_index
389        .ok_or_else(|| TransformError::missing_metadata("video.progress.child"))?;
390    let result = progress
391        .result
392        .clone()
393        .ok_or_else(|| TransformError::missing_metadata("video.progress.result"))?;
394    if progress
395        .state
396        .as_ref()
397        .is_none_or(|s| &s.binding != binding)
398    {
399        return Err(TransformError::shape(
400            "video.progress.binding",
401            "different saved invocation",
402        ));
403    }
404    let (mut state, version) = load(store, scope, binding, limits).await?;
405    if state.children.get(index).is_none_or(|c| !c.started) {
406        return Err(conflict("video.child.reservation"));
407    }
408
409    state.children[index].result = Some(result);
410    state.children[index].published = None;
411    bound_value(&state, limits)?;
412    progress.version = Some(version);
413    save(store, scope, &state, progress, limits).await
414}
415
416fn copy_template(request: &WireRequest<()>) -> WireRequest<()> {
417    WireRequest {
418        method: request.method.clone(),
419        path: request.path.clone(),
420        query: request.query.clone(),
421        headers: request.headers.clone(),
422        body: (),
423    }
424}
gproxy_protocol/adapt/video/reverse/invoke.rs

gproxy_protocol/adapt/video/reverse/
invoke.rs
