sofie_db/worker/
context.rs

1use std::sync::Arc;
2
3use serde_json::Value;
4use sofie_auth::SigningKeys;
5use sofie_core::{Error, Result, Uuid};
6
7use crate::{Claimed, Context, Db, Outcome, Reach, Step, Writes};
8
9pub struct JobContext {
10    pub(crate) db: Db,
11    pub(crate) payload: Value,
12    event: Option<&'static str>,
13    actor: Option<Uuid>,
14    completion: Step,
15    #[cfg(feature = "remote-contracts")]
16    pub(crate) message: sofie_core::Uuid,
17    /// Keyed by the message's id, so a retry gets the same key.
18    #[cfg(feature = "idempotency")]
19    pub(crate) work: crate::Work,
20    #[cfg(feature = "contracts")]
21    pub(crate) retried_for: std::time::Duration,
22}
23
24impl JobContext {
25    pub(super) fn new(
26        worker: &super::Worker,
27        message: &Claimed,
28        event: Option<&'static str>,
29        completion: Step,
30    ) -> Self {
31        Self {
32            db: worker.db.clone(),
33            payload: message.payload.clone(),
34            event,
35            actor: message.actor,
36            completion,
37            #[cfg(feature = "remote-contracts")]
38            message: message.id,
39            #[cfg(feature = "idempotency")]
40            work: crate::Work::new(message.id.to_string(), message.name.clone()),
41            #[cfg(feature = "contracts")]
42            retried_for: super::backoff::retried_for(worker.max_attempts),
43        }
44    }
45
46    /// The keys requests sign links with, so a job's links open the same routes.
47    pub fn signing_keys(&self) -> Result<Arc<SigningKeys>> {
48        self.db.signing_keys().ok_or_else(|| {
49            Error::internal_message(
50                "the worker's database has no signing keys; give it the app's with \
51                 `Db::with_signing_keys`",
52            )
53        })
54    }
55
56    pub(crate) async fn complete(self, outcome: &mut dyn Outcome) -> Result<()> {
57        let reach = self.reach();
58        self.db
59            .commit_with(outcome, vec![self.completion], reach)
60            .await
61    }
62}
63
64/// A job reaches every record, as the principal whose action queued it if one did.
65impl Context for JobContext {
66    fn db(&self) -> &Db {
67        &self.db
68    }
69
70    fn reach(&self) -> Reach {
71        self.actor.map_or(Reach::All, Reach::AllAs)
72    }
73
74    /// The framework wrote the payload, so a missing id is its fault, not a lookup's.
75    fn bound(&self, name: &str) -> Result<Uuid> {
76        self.payload
77            .get(name)
78            .and_then(Value::as_str)
79            .and_then(|text| text.parse().ok())
80            .ok_or_else(|| Error::internal_message(format!("the payload has no id named `{name}`")))
81    }
82
83    #[cfg(feature = "jobs")]
84    fn input(&mut self) -> Result<Value> {
85        Ok(self.payload.clone())
86    }
87
88    fn event(&self, name: &str) -> Result<Value> {
89        match self.event {
90            Some(delivered) if delivered == name => Ok(self.payload.clone()),
91            Some(delivered) => Err(Error::internal_message(format!(
92                "this listener receives `{delivered}`, so it cannot take a `{name}`"
93            ))),
94            None => Err(Error::internal_message(format!(
95                "a scheduled job receives no event, so it cannot take a `{name}`"
96            ))),
97        }
98    }
99}
100
101impl Writes for JobContext {}
sofie_db/worker/context.rs

sofie_db/worker/
context.rs
