sofie_client/
remote.rs

1//! The calling side of a split: contracts, batch lookups and the event feed of another app.
2
3use std::collections::BTreeMap;
4use std::marker::PhantomData;
5use std::sync::Arc;
6use std::time::Duration;
7
8use serde::Deserialize;
9use serde::de::DeserializeOwned;
10use serde_json::Value;
11use sofie_auth::SigningKeys;
12use sofie_core::{Error, FieldError, Result, Timestamp, Uuid};
13use sofie_db::remote::{CONTRACTS_PATH, FEED_PATH, LOOKUPS_PATH, PRINCIPAL_HEADER, lookup_name};
14use sofie_db::{BoxFuture, FeedEntry, FeedSource, Lookup};
15use tracing::Instrument;
16
17use crate::{Outgoing, Request, Response};
18
19#[derive(Clone)]
20pub struct RemoteApp {
21    shared: Arc<Connection>,
22}
23
24struct Connection {
25    name: &'static str,
26    outgoing: Outgoing,
27    keys: SigningKeys,
28}
29
30impl RemoteApp {
31    /// The wiring's constructor: `name` is the split module, whose keys name the app.
32    #[doc(hidden)]
33    #[must_use]
34    pub fn new(name: &'static str, outgoing: Outgoing, keys: SigningKeys) -> Self {
35        Self {
36            shared: Arc::new(Connection {
37                name,
38                outgoing,
39                keys,
40            }),
41        }
42    }
43
44    #[must_use]
45    pub fn name(&self) -> &'static str {
46        self.shared.name
47    }
48
49    async fn post<R: DeserializeOwned>(&self, path: &str, arguments: Result<Value>) -> Result<R> {
50        let request = self.shared.outgoing.post(path).json(&arguments?);
51        self.send(request, path).await
52    }
53
54    /// The principal token carries the principal the current work acts for, valid for `path` only.
55    async fn send<R: DeserializeOwned>(&self, request: Request<'_>, path: &str) -> Result<R> {
56        let acting = sofie_db::remote::acting();
57        let token = self
58            .shared
59            .keys
60            .principal_token(path, acting.as_ref(), Timestamp::now());
61        let response = request.header(PRINCIPAL_HEADER, &token).send().await?;
62        if response.status() == 200 {
63            return response.json();
64        }
65        Err(refused(self.shared.name, &response))
66    }
67}
68
69/// `{"error": …}`, or `{"errors": [{"field", "key", "arguments", "message"}]}` for invalid input.
70#[derive(Default, Deserialize)]
71#[serde(default)]
72struct Refusal {
73    error: String,
74    errors: Option<Vec<RefusedField>>,
75}
76
77/// Keyed so the calling app translates the refusal as if it had raised it.
78#[derive(Deserialize)]
79struct RefusedField {
80    field: String,
81    /// An app deployed before keys travelled sends none; its message then stands as written.
82    #[serde(default = "unkeyed")]
83    key: String,
84    #[serde(default)]
85    arguments: BTreeMap<String, String>,
86    message: String,
87}
88
89/// The error the serving app answered with, so a caller handles it as an in-process one.
90fn refused(app: &str, response: &Response) -> Error {
91    let refusal: Refusal = serde_json::from_slice(response.body()).unwrap_or_default();
92    match response.status() {
93        400 => Error::BadRequest(refusal.error),
94        404 => Error::NotFound,
95        409 => Error::Conflict,
96        422 => match refusal.errors {
97            Some(fields) => Error::Invalid(
98                fields
99                    .into_iter()
100                    .map(RefusedField::into_field_error)
101                    .collect(),
102            ),
103            None => Error::Rule(refusal.error),
104        },
105        429 => Error::TooManyRequests(retry_after(response)),
106        503 => Error::Unavailable(format!("{app} is unavailable")),
107        status => Error::internal_message(format!(
108            "{app} answered {status}; its own trace holds the reason"
109        )),
110    }
111}
112
113fn unkeyed() -> String {
114    "remote".to_owned()
115}
116
117impl RefusedField {
118    fn into_field_error(self) -> FieldError {
119        self.arguments.into_iter().fold(
120            FieldError::new(self.field, self.key, self.message),
121            |error, (name, value)| error.with(name, value),
122        )
123    }
124}
125
126const MONTHS: [&str; 12] = [
127    "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec",
128];
129
130/// Seconds, or an HTTP date (`Sun, 06 Nov 1994 08:49:37 GMT`) after which to try again.
131fn retry_after(response: &Response) -> Duration {
132    let Some(value) = response.header("retry-after") else {
133        return Duration::from_secs(1);
134    };
135    if let Ok(seconds) = value.trim().parse() {
136        return Duration::from_secs(seconds);
137    }
138    http_date(value).map_or(Duration::from_secs(1), |at| {
139        Duration::try_from(at.duration_since(Timestamp::now())).unwrap_or(Duration::ZERO)
140    })
141}
142
143fn http_date(text: &str) -> Option<Timestamp> {
144    let words: Vec<&str> = text.split_whitespace().collect();
145    let [_weekday, day, month, year, time, "GMT"] = words.as_slice() else {
146        return None;
147    };
148    let month = MONTHS.iter().position(|name| name == month)? + 1;
149    format!("{year}-{month:02}-{day}T{time}Z").parse().ok()
150}
151
152/// `#[contract]` implements each contract trait `T` for `Remote<dyn T>`, calling these.
153pub struct Remote<T: ?Sized> {
154    app: RemoteApp,
155    contract: PhantomData<fn() -> Box<T>>,
156}
157
158impl<T: ?Sized> Remote<T> {
159    #[must_use]
160    pub fn new(app: RemoteApp) -> Self {
161        Self {
162            app,
163            contract: PhantomData,
164        }
165    }
166
167    /// `contract` is `<contract crate>/<trait>`; the arguments are serialized before the call.
168    pub fn call<'a, R: DeserializeOwned + Send + 'a>(
169        &'a self,
170        contract: &'static str,
171        method: &'static str,
172        arguments: Result<Value>,
173    ) -> BoxFuture<'a, Result<R>> {
174        let name = contract.rsplit_once('/').map_or(contract, |(_, name)| name);
175        let span = tracing::info_span!(
176            "contract",
177            otel.name = format!("{name}::{method}"),
178            contract.app = self.app.name(),
179        );
180        let path = format!("{CONTRACTS_PATH}{contract}/{method}");
181        Box::pin(async move { self.app.post(&path, arguments).await }.instrument(span))
182    }
183}
184
185pub struct RemoteLookup<D> {
186    app: RemoteApp,
187    dto: PhantomData<fn() -> D>,
188}
189
190impl<D> RemoteLookup<D> {
191    #[must_use]
192    pub fn new(app: RemoteApp) -> Self {
193        Self {
194            app,
195            dto: PhantomData,
196        }
197    }
198}
199
200impl<D: DeserializeOwned + Send> Lookup<D> for RemoteLookup<D> {
201    fn get_many<'a>(&'a self, ids: &'a [Uuid]) -> BoxFuture<'a, Result<Vec<D>>> {
202        let path = format!("{LOOKUPS_PATH}{}", lookup_name::<D>());
203        let ids = serde_json::to_value(ids)
204            .map_err(|error| Error::internal("serializing the ids of a lookup", error));
205        Box::pin(async move {
206            let request = self.app.shared.outgoing.post(&path).json(&ids?);
207            self.app.send(request.repeatable(), &path).await
208        })
209    }
210}
211
212pub struct RemoteFeed {
213    app: RemoteApp,
214}
215
216impl RemoteFeed {
217    #[must_use]
218    pub fn new(app: RemoteApp) -> Self {
219        Self { app }
220    }
221}
222
223impl FeedSource for RemoteFeed {
224    fn producer(&self) -> &str {
225        self.app.name()
226    }
227
228    fn after(&self, position: i64, limit: i64) -> BoxFuture<'_, Result<Vec<FeedEntry>>> {
229        let page = format!("{FEED_PATH}?after={position}&limit={limit}");
230        Box::pin(async move {
231            let request = self.app.shared.outgoing.get(&page);
232            self.app.send(request, FEED_PATH).await
233        })
234    }
235}
236
237#[cfg(test)]
238mod tests {
239    use bytes::Bytes;
240    use http::{HeaderMap, StatusCode};
241
242    use super::*;
243
244    fn answered(status: u16, body: &str) -> Error {
245        let response = Response {
246            vendor: "billing",
247            status: StatusCode::from_u16(status).unwrap(),
248            headers: HeaderMap::new(),
249            body: Bytes::from(body.to_owned()),
250        };
251        refused("billing", &response)
252    }
253
254    #[test]
255    fn a_refusal_is_read_back_into_the_error_the_serving_app_raised() {
256        let invalid = answered(
257            422,
258            r#"{"errors":[{"field":"email","key":"validation.unique","arguments":{},"message":"is taken"}]}"#,
259        );
260        let Error::Invalid(fields) = invalid else {
261            panic!("invalid input: {invalid:?}");
262        };
263
264        assert_eq!(
265            fields,
266            [FieldError::new("email", "validation.unique", "is taken")]
267        );
268        assert!(
269            matches!(answered(422, r#"{"error":"closed"}"#), Error::Rule(message) if message == "closed")
270        );
271        assert!(
272            matches!(answered(400, "not json"), Error::BadRequest(message) if message.is_empty())
273        );
274        assert!(matches!(answered(404, ""), Error::NotFound));
275    }
276
277    #[test]
278    fn a_refused_field_keeps_its_key_and_arguments() {
279        let invalid = answered(
280            422,
281            r#"{"errors":[{"field":"code","key":"validation.max_length","arguments":{"max":"4"},"message":"must be at most 4"}]}"#,
282        );
283        let Error::Invalid(fields) = invalid else {
284            panic!("invalid input: {invalid:?}");
285        };
286
287        assert_eq!(
288            fields,
289            [
290                FieldError::new("code", "validation.max_length", "must be at most 4")
291                    .with("max", "4")
292            ]
293        );
294    }
295
296    #[test]
297    fn a_refused_field_from_an_app_that_sends_no_key_keeps_its_message() {
298        let invalid = answered(
299            422,
300            r#"{"errors":[{"field":"email","message":"is taken"}]}"#,
301        );
302        let Error::Invalid(fields) = invalid else {
303            panic!("invalid input: {invalid:?}");
304        };
305
306        assert_eq!(fields, [FieldError::new("email", "remote", "is taken")]);
307    }
308
309    fn told_to_wait(retry_after: &str) -> Duration {
310        let mut headers = HeaderMap::new();
311        headers.insert("retry-after", retry_after.parse().unwrap());
312        let response = Response {
313            vendor: "billing",
314            status: StatusCode::TOO_MANY_REQUESTS,
315            headers,
316            body: Bytes::new(),
317        };
318        let Error::TooManyRequests(wait) = refused("billing", &response) else {
319            panic!("a 429 is too many requests");
320        };
321        wait
322    }
323
324    #[test]
325    fn retry_after_is_read_as_seconds_or_as_a_date() {
326        let in_two_minutes = Timestamp::now() + Duration::from_secs(120);
327        let date = in_two_minutes
328            .strftime("%a, %d %b %Y %H:%M:%S GMT")
329            .to_string();
330
331        let waited = told_to_wait(&date);
332
333        assert!(
334            (Duration::from_secs(100)..=Duration::from_secs(120)).contains(&waited),
335            "{waited:?}"
336        );
337        assert_eq!(told_to_wait("7"), Duration::from_secs(7));
338        assert_eq!(
339            told_to_wait("Sun, 06 Nov 1994 08:49:37 GMT"),
340            Duration::ZERO
341        );
342        assert_eq!(told_to_wait("soon"), Duration::from_secs(1));
343    }
344}
sofie_client/remote.rs

sofie_client/
remote.rs
