sofie_client/
remote.rs

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

sofie_client/
remote.rs
