gproxy_protocol/adapt/video/reverse/
state.rs

1use super::*;
2use serde::{Deserialize, Serialize};
3
4#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
5pub enum ReverseVideoKind {
6    Native,
7    OpenRouter,
8}
9
10/// Host-selected target/API facts, persisted before any publication or create.
11#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
12pub struct ReverseVideoBinding {
13    pub operation_name: String,
14    pub origin: String,
15    pub model: String,
16    pub kind: ReverseVideoKind,
17    pub api_origin_url: String,
18    pub create_path: String,
19    pub query_prefix: String,
20}
21
22#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
23pub enum ReverseVideoRequest {
24    Native(o::NativeCreateVideoRequestBody),
25    OpenRouter(o::CreateVideoRequestBody),
26}
27
28#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
29pub enum ReverseVideoResult {
30    Native(o::NativeVideo),
31    OpenRouter(o::VideoGenerationResponseBody),
32}
33
34#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
35pub struct ReverseVideoChild {
36    pub instance_index: usize,
37    pub sample_index: usize,
38    pub source: g::PredictLongRunningRequestBody,
39    pub request: Option<ReverseVideoRequest>,
40    /// Persisted before POST. Absence of a result never permits another POST.
41    pub started: bool,
42    pub result: Option<ReverseVideoResult>,
43    pub published: Option<g::VideoOperation>,
44}
45
46#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
47pub struct ReverseVideoState {
48    pub schema: u32,
49    pub native_defaults: Option<video::NativeVideoDefaults>,
50    pub expires_at: SystemTime,
51    pub binding: ReverseVideoBinding,
52    pub original: g::PredictLongRunningRequestBody,
53    pub children: Vec<ReverseVideoChild>,
54}
55
56#[derive(Debug, Default)]
57pub struct ReverseVideoProgress {
58    pub state: Option<ReverseVideoState>,
59    pub version: Option<Version>,
60    pub child_index: Option<usize>,
61    pub send_started: bool,
62    pub raw_response: Option<WireResponse<bytes::Bytes>>,
63    pub result: Option<ReverseVideoResult>,
64    pub publication_ids: Vec<String>,
65    pub publications: Vec<ResourceReference>,
66}
67
68pub(super) fn conflict(path: &str) -> TransformError {
69    TransformError::new(
70        TransformErrorKind::Conflict,
71        path,
72        "saved creation requires reconciliation; do not repeat POST",
73    )
74}
75
76impl ReverseVideoBinding {
77    pub(super) fn key(&self) -> String {
78        format!(
79            "video/reverse/{}:{}",
80            self.operation_name.len(),
81            self.operation_name
82        )
83    }
84
85    pub(super) fn query_path(&self, id: &str) -> Result<String, TransformError> {
86        if id.is_empty() || id.chars().any(char::is_control) {
87            return Err(TransformError::shape(
88                "video.id",
89                "nonempty control-free target identity required",
90            ));
91        }
92        let mut encoded = String::new();
93        for byte in id.bytes() {
94            if byte.is_ascii_alphanumeric()
95                || matches!(byte, b'-' | b'_' | b'~')
96                || byte == b'.' && !matches!(id, "." | "..")
97            {
98                encoded.push(char::from(byte));
99            } else {
100                use std::fmt::Write;
101                write!(&mut encoded, "%{byte:02X}").expect("String write");
102            }
103        }
104        Ok(format!("{}{encoded}", self.query_prefix))
105    }
106}
107
108impl ReverseVideoResult {
109    pub(super) fn id(&self) -> &str {
110        match self {
111            Self::Native(v) => &v.id,
112            Self::OpenRouter(v) => &v.id,
113        }
114    }
115    pub(super) fn same_projection(&self, other: &Self) -> bool {
116        match (self, other) {
117            (Self::Native(a), Self::Native(b)) => {
118                a.id == b.id && a.status == b.status && a.error == b.error
119            }
120            (Self::OpenRouter(a), Self::OpenRouter(b)) => {
121                a.id == b.id
122                    && a.status == b.status
123                    && a.error == b.error
124                    && a.unsigned_urls == b.unsigned_urls
125            }
126            _ => false,
127        }
128    }
129    pub(super) fn clean(self) -> Self {
130        match self {
131            Self::Native(v) => Self::Native(v.into_declared()),
132            Self::OpenRouter(v) => Self::OpenRouter(v.into_declared()),
133        }
134    }
135}
136
137pub(super) async fn save<S: StateStore>(
138    store: &S,
139    scope: &S::Scope,
140    state: &ReverseVideoState,
141    progress: &mut ReverseVideoProgress,
142    limits: VideoLimits,
143) -> Result<(), TransformError> {
144    // Retain the known outcome even when CAS fails after a side effect.
145    progress.state = Some(state.clone());
146    let version = super::super::state::save_payload(
147        store,
148        scope,
149        &state.binding.key(),
150        state,
151        progress.version.clone(),
152        super::super::state::state_expiry(Some(state.expires_at)),
153        limits,
154    )
155    .await?;
156    progress.version = Some(version);
157    Ok(())
158}
159
160pub(super) async fn load<S: StateStore>(
161    store: &S,
162    scope: &S::Scope,
163    binding: &ReverseVideoBinding,
164    limits: VideoLimits,
165) -> Result<(ReverseVideoState, Version), TransformError> {
166    let entry = store
167        .get(scope, &binding.key())
168        .await?
169        .ok_or_else(|| TransformError::missing_metadata("video.reverse.state"))?;
170    if entry.payload.len() as u64 > store.limits().read_bytes {
171        return Err(super::super::resources::limit("video.state"));
172    }
173    let mut state: ReverseVideoState = crate::codec::decode_json(&entry.payload, limits.codec)
174        .map_err(|e| TransformError::invalid_result("video.state", e.to_string()))?;
175    if state.schema != 1 || &state.binding != binding || state.children.is_empty() {
176        return Err(TransformError::shape(
177            "video.state.binding",
178            "invalid schema, binding or child count",
179        ));
180    }
181    if state
182        .children
183        .iter()
184        .any(|child| child.source.instances.len() != 1)
185    {
186        return Err(TransformError::invalid_result(
187            "video.state",
188            "each child must retain exactly one source instance",
189        ));
190    }
191    if state.expires_at <= SystemTime::now() {
192        return Err(TransformError::missing_metadata(
193            "video.state.unexpired_publications",
194        ));
195    }
196    state.original = state.original.into_declared();
197    for child in &mut state.children {
198        child.source = std::mem::replace(
199            &mut child.source,
200            g::PredictLongRunningRequestBody::builder(vec![]).build(),
201        )
202        .into_declared();
203        child.request = child.request.take().map(|v| match v {
204            ReverseVideoRequest::Native(v) => ReverseVideoRequest::Native(v.into_declared()),
205            ReverseVideoRequest::OpenRouter(v) => {
206                ReverseVideoRequest::OpenRouter(v.into_declared())
207            }
208        });
209        child.result = child.result.take().map(ReverseVideoResult::clean);
210        child.published = child.published.take().map(DeclaredFields::into_declared);
211    }
212    Ok((state, entry.version))
213}
gproxy_protocol/adapt/video/reverse/state.rs

gproxy_protocol/adapt/video/reverse/
state.rs
