gproxy_protocol/adapt/video/
native.rs

1use super::*;
2use super::{
3    composed::{operation, prepare_headers},
4    resources::*,
5    state::*,
6};
7use crate::{
8    HttpBody, WireResponse,
9    capability::{
10        ResourceAccess, ResourceMetadata, ResourceRead, ResourceReference, StateStore, Version,
11    },
12    transform::{
13        TransformErrorKind,
14        video::{NativeVideoDefaults, NativeVideoResponseFacts},
15    },
16};
17use base64::{Engine as _, engine::general_purpose::STANDARD};
18
19async fn input_resources<R: ResourceAccess>(
20    access: &R,
21    scope: &R::Scope,
22    input: &o::NativeCreateVideoRequestBody,
23    limits: VideoLimits,
24) -> Result<BTreeMap<String, ResolvedVideoResource>, TransformError> {
25    let mut out = BTreeMap::new();
26    let Some(reference) = &input.input_reference else {
27        return Ok(out);
28    };
29
30    let (key, reference) = match reference {
31        o::NativeInputReference::File(v) => {
32            (v.file_id.clone(), ResourceReference::Id(v.file_id.clone()))
33        }
34        o::NativeInputReference::Url(v) => (
35            v.image_url.clone(),
36            ResourceReference::Url(v.image_url.clone()),
37        ),
38    };
39    let (bytes, mime) = read(
40        access,
41        scope,
42        &reference,
43        true,
44        limits,
45        limits.codec.max_body_bytes,
46    )
47    .await?;
48    out.insert(
49        key.clone(),
50        ResolvedVideoResource {
51            reference: key,
52            url: None,
53            bytes_base64_encoded: Some(STANDARD.encode(bytes)),
54            mime_type: Some(mime),
55        },
56    );
57    Ok(out)
58}
59
60#[allow(clippy::too_many_arguments)]
61pub async fn native_to_gemini_create_composed<
62    U: Upstream,
63    R: ResourceAccess,
64    S: StateStore,
65    F: FnOnce(&g::VideoOperation) -> Result<NativeVideoResponseFacts, TransformError>,
66>(
67    upstream: &U,
68    target: &U::Target,
69    access: &R,
70    source_resource_scope: &R::Scope,
71    store: &S,
72    state_scope: &S::Scope,
73    template: WireRequest<()>,
74    input: o::NativeCreateVideoRequestBody,
75    defaults: Option<NativeVideoDefaults>,
76    binding: VideoBinding,
77    facts: F,
78    progress: &mut VideoProgress,
79    limits: VideoLimits,
80) -> Result<JsonInvocation<Converted<o::NativeVideo>>, TransformError> {
81    super::composed::fresh(progress)?;
82
83    direct::check_template(&template, http::Method::POST)?;
84    if template.path != binding.create_path()? {
85        return Err(TransformError::shape(
86            "video.create.path",
87            "path differs from selected Veo model/API prefix",
88        ));
89    }
90    let input = input.into_declared();
91    bound_value(&input, limits)?;
92    bound_value(&binding, limits)?;
93    let resources = input_resources(access, source_resource_scope, &input, limits).await?;
94    let prepared =
95        video::native_to_gemini_request(input.clone(), &binding.model, defaults, &resources)?;
96    let mut codec = limits.codec;
97    codec.max_body_bytes = codec.max_body_bytes.min(upstream.limits().write_bytes);
98    codec.max_buffer_bytes = codec.max_buffer_bytes.min(codec.max_body_bytes);
99    codec.max_value_bytes = codec.max_value_bytes.min(codec.max_body_bytes);
100    let bytes = crate::codec::encode_json(&prepared.value.body, codec).map_err(|e| {
101        TransformError::new(TransformErrorKind::Limit, "video.request", e.to_string())
102    })?;
103    let mut state = VideoJobState {
104        schema: 1,
105        binding,
106        original: VideoOriginalRequest::Native(input),
107        native_facts: None,
108        effective: prepared.value.body,
109        operation: None,
110        published_urls: BTreeMap::new(),
111    };
112    progress.binding = Some(state.binding.clone());
113    let version = save(store, state_scope, &state, None, limits).await?;
114    progress.reserved = true;
115    let response = super::transport::invoke(
116        upstream,
117        target,
118        prepare_headers(template, bytes, true),
119        &mut progress.send_started,
120        &mut progress.raw_response,
121        &mut progress.native_response,
122        limits,
123    )
124    .await?;
125    let response = match response {
126        JsonInvocation::Success(response) => response,
127        JsonInvocation::Rejected(r) => return Ok(JsonInvocation::Rejected(r)),
128    };
129    operation(&response.body, &state.binding)?;
130    state.operation = Some(response.body.clone());
131    let version = save(store, state_scope, &state, Some(version), limits).await?;
132    progress.binding_saved = true;
133    finish(
134        store,
135        state_scope,
136        response,
137        &mut state,
138        version,
139        facts,
140        progress,
141        limits,
142    )
143    .await
144}
145
146#[allow(clippy::too_many_arguments)]
147async fn finish<
148    S: StateStore,
149    F: FnOnce(&g::VideoOperation) -> Result<NativeVideoResponseFacts, TransformError>,
150>(
151    store: &S,
152    scope: &S::Scope,
153    response: WireResponse<g::VideoOperation>,
154    state: &mut VideoJobState,
155    version: Version,
156    facts: F,
157    progress: &mut VideoProgress,
158    limits: VideoLimits,
159) -> Result<JsonInvocation<Converted<o::NativeVideo>>, TransformError> {
160    let facts = facts(&response.body)?;
161    bound_value(&facts, limits)?;
162    let VideoOriginalRequest::Native(original) = &state.original else {
163        return Err(TransformError::shape(
164            "video.source",
165            "native source job required",
166        ));
167    };
168    if facts.client_id != state.binding.client_id
169        || original.model.as_ref().is_some_and(|m| m != &facts.model)
170    {
171        return Err(TransformError::shape(
172            "native.facts.identity",
173            "native response facts differ from original client identity/model",
174        ));
175    }
176    let expected = state
177        .effective
178        .parameters
179        .as_ref()
180        .ok_or_else(|| TransformError::missing_metadata("native.effective_parameters"))?;
181    let seconds = match facts.seconds {
182        o::NativeVideoSeconds::Four => 4,
183        o::NativeVideoSeconds::Eight => 8,
184        o::NativeVideoSeconds::Twelve => 12,
185    };
186    let aspect = match facts.size {
187        o::NativeVideoSize::Portrait720 => "9:16",
188        o::NativeVideoSize::Landscape720 => "16:9",
189        _ => {
190            return Err(TransformError::shape(
191                "native.facts.size",
192                "result differs from exact Veo dimension mapping",
193            ));
194        }
195    };
196    if expected.duration_seconds != Some(seconds)
197        || expected.aspect_ratio.as_deref() != Some(aspect)
198        || expected.resolution.as_deref() != Some("720p")
199        || facts.prompt.as_ref().is_some_and(|p| p != &original.prompt)
200    {
201        return Err(TransformError::shape(
202            "native.facts.request",
203            "native output facts contradict the effective request",
204        ));
205    }
206    if state.native_facts.as_ref().is_some_and(|old| {
207        old.created_at != facts.created_at
208            || old.model != facts.model
209            || old.seconds != facts.seconds
210            || old.size != facts.size
211            || old.progress > facts.progress
212    }) {
213        return Err(TransformError::shape(
214            "native.facts.lifecycle",
215            "stable native metadata changed or progress decreased",
216        ));
217    }
218    let converted = video::gemini_operation_to_native(response.body, &facts)?;
219    bound_value(&converted.value.body, limits)?;
220    state.native_facts = Some(facts);
221    save(store, scope, state, Some(version), limits).await?;
222    progress.binding_saved = true;
223    Ok(JsonInvocation::Success(WireResponse {
224        status: response.status,
225        headers: response.headers,
226        body: Converted {
227            value: converted.value.body,
228            report: converted.report,
229        },
230    }))
231}
232
233#[allow(clippy::too_many_arguments)]
234pub async fn native_to_gemini_query_composed<
235    U: Upstream,
236    S: StateStore,
237    F: FnOnce(&g::VideoOperation) -> Result<NativeVideoResponseFacts, TransformError>,
238>(
239    upstream: &U,
240    target: &U::Target,
241    store: &S,
242    state_scope: &S::Scope,
243    template: WireRequest<()>,
244    binding: VideoBinding,
245    facts: F,
246    progress: &mut VideoProgress,
247    limits: VideoLimits,
248) -> Result<JsonInvocation<Converted<o::NativeVideo>>, TransformError> {
249    super::composed::fresh(progress)?;
250
251    direct::check_template(&template, http::Method::GET)?;
252    let (mut state, version) = load(store, state_scope, &binding, limits).await?;
253    if !matches!(state.original, VideoOriginalRequest::Native(_)) {
254        return Err(TransformError::shape(
255            "video.source",
256            "native source job required",
257        ));
258    }
259    let old = state
260        .operation
261        .as_ref()
262        .ok_or_else(|| TransformError::missing_metadata("native.uncertain_operation"))?;
263    let name = operation(old, &binding)?.to_owned();
264    if template.path != binding.query_path(&name)? {
265        return Err(TransformError::shape(
266            "native.query.path",
267            "path differs from stored operation",
268        ));
269    }
270    progress.binding = Some(binding.clone());
271    let response = super::transport::invoke(
272        upstream,
273        target,
274        prepare_headers(template, bytes::Bytes::new(), false),
275        &mut progress.send_started,
276        &mut progress.raw_response,
277        &mut progress.native_response,
278        limits,
279    )
280    .await?;
281    let response = match response {
282        JsonInvocation::Success(response) => response,
283        JsonInvocation::Rejected(r) => return Ok(JsonInvocation::Rejected(r)),
284    };
285    if operation(&response.body, &binding)? != name
286        || old.done == Some(true)
287            && (old.done != response.body.done
288                || old.response != response.body.response
289                || old.error != response.body.error)
290    {
291        return Err(TransformError::invalid_result(
292            "native.query.lifecycle",
293            "operation identity or terminal result changed",
294        ));
295    }
296    state.operation = Some(response.body.clone());
297    let version = save(store, state_scope, &state, Some(version), limits).await?;
298    progress.binding_saved = true;
299    finish(
300        store,
301        state_scope,
302        response,
303        &mut state,
304        version,
305        facts,
306        progress,
307        limits,
308    )
309    .await
310}
311
312#[allow(clippy::too_many_arguments)]
313pub async fn recover_native_result<
314    S: StateStore,
315    F: FnOnce(&g::VideoOperation) -> Result<NativeVideoResponseFacts, TransformError>,
316>(
317    store: &S,
318    scope: &S::Scope,
319    binding: VideoBinding,
320    facts: F,
321    progress: &mut VideoProgress,
322    limits: VideoLimits,
323) -> Result<JsonInvocation<Converted<o::NativeVideo>>, TransformError> {
324    if progress.binding.as_ref() != Some(&binding) || !progress.send_started {
325        return Err(TransformError::shape(
326            "native.recovery",
327            "progress belongs to a different invocation",
328        ));
329    }
330    let native = progress
331        .native_response
332        .as_ref()
333        .ok_or_else(|| TransformError::missing_metadata("native.recovery.response"))?;
334    let response = WireResponse {
335        status: native.status,
336        headers: native.headers.clone(),
337        body: native.body.clone(),
338    };
339    operation(&response.body, &binding)?;
340    let (mut state, version) = load(store, scope, &binding, limits).await?;
341    if state
342        .operation
343        .as_ref()
344        .is_some_and(|old| old != &response.body)
345    {
346        return Err(TransformError::new(
347            TransformErrorKind::Conflict,
348            "native.recovery",
349            "stored operation differs from retained source",
350        ));
351    }
352    state.operation = Some(response.body.clone());
353    let version = save(store, scope, &state, Some(version), limits).await?;
354    finish(
355        store, scope, response, &mut state, version, facts, progress, limits,
356    )
357    .await
358}
359
360/// Reads the actual completed video's bytes for the native `/content` route.
361/// The selected target resource scope supplies authorization for private URIs.
362pub async fn native_content<R: ResourceAccess, S: StateStore>(
363    access: &R,
364    target_resource_scope: &R::Scope,
365    store: &S,
366    state_scope: &S::Scope,
367    binding: VideoBinding,
368    limits: VideoLimits,
369) -> Result<ResourceRead, TransformError> {
370    let (state, _) = load(store, state_scope, &binding, limits).await?;
371    if !matches!(state.original, VideoOriginalRequest::Native(_)) {
372        return Err(TransformError::shape(
373            "video.source",
374            "native source job required",
375        ));
376    }
377    let facts = state
378        .native_facts
379        .ok_or_else(|| TransformError::missing_metadata("native.result_facts"))?;
380    let expires_at = facts
381        .expires_at
382        .map(|seconds| {
383            std::time::UNIX_EPOCH
384                .checked_add(std::time::Duration::from_secs(seconds as u64))
385                .ok_or_else(|| TransformError::shape("native.expires_at", "timestamp overflow"))
386        })
387        .transpose()?;
388    if expires_at.is_some_and(|v| v <= std::time::SystemTime::now()) {
389        return Err(TransformError::new(
390            TransformErrorKind::MissingMetadata,
391            "native.content",
392            "saved video result has expired",
393        ));
394    }
395    let result = video::gemini_operation_to_native(
396        state
397            .operation
398            .ok_or_else(|| TransformError::missing_metadata("native.operation"))?,
399        &facts,
400    )?;
401    if result.value.body.status != o::NativeVideoStatus::Completed {
402        return Err(TransformError::new(
403            TransformErrorKind::Conflict,
404            "native.content",
405            "video is not completed",
406        ));
407    }
408    let video = result
409        .value
410        .videos
411        .into_iter()
412        .next()
413        .ok_or_else(|| TransformError::missing_metadata("native.content"))?;
414    let (bytes, mime) = if let Some(uri) = video.uri {
415        read(
416            access,
417            target_resource_scope,
418            &ResourceReference::Url(uri),
419            false,
420            limits,
421            limits.codec.max_body_bytes,
422        )
423        .await?
424    } else {
425        let mime = video
426            .encoding
427            .ok_or_else(|| TransformError::missing_metadata("native.content.mime"))?;
428        let bytes = decoded(
429            video
430                .encoded_video
431                .as_deref()
432                .ok_or_else(|| TransformError::missing_metadata("native.content.bytes"))?,
433            &mime,
434            false,
435            limits.codec.max_body_bytes,
436        )?;
437        (bytes, mime)
438    };
439    Ok(ResourceRead {
440        metadata: ResourceMetadata {
441            mime: Some(mime),
442            length: Some(bytes.len() as u64),
443            filename: Some("video.mp4".into()),
444            expires_at,
445        },
446        body: HttpBody::Bytes(bytes),
447    })
448}
gproxy_protocol/adapt/video/native.rs

gproxy_protocol/adapt/video/
native.rs
