runledger_runtime/catalog/
registration.rs

1use std::collections::BTreeMap;
2use std::sync::Arc;
3
4use runledger_core::jobs::{JobHandler, JobHandlerRegistry, JobType, JobTypeName};
5use runledger_postgres::jobs::JobDefinitionUpsert;
6
7use crate::registry::JobRegistry;
8
9use super::schedule_spec::{CatalogJobScheduleSpec, StoredCatalogJobScheduleSpec};
10use super::types::CatalogJob;
11use super::{
12    CatalogError, JobCatalog, JobCatalogDefaults, JobCatalogDefinitionOverrides,
13    JobCatalogSyncScope,
14};
15
16impl JobCatalog {
17    /// Creates an empty catalog with default definition values.
18    #[must_use]
19    pub fn new() -> Self {
20        Self {
21            defaults: JobCatalogDefaults::default(),
22            jobs: BTreeMap::new(),
23            schedules: Vec::new(),
24        }
25    }
26
27    /// Replaces the definition defaults used by subsequent sync operations.
28    #[must_use]
29    pub fn defaults(mut self, defaults: JobCatalogDefaults) -> Self {
30        self.defaults = defaults;
31        self
32    }
33
34    /// Registers a handler after validating declared and handler job types match.
35    ///
36    /// # Errors
37    /// Returns [`CatalogError`] when job types are blank, mismatched, or duplicated.
38    pub fn try_job<H>(mut self, job_type: &'static str, handler: H) -> Result<Self, CatalogError>
39    where
40        H: JobHandler + 'static,
41    {
42        let declared = Self::validate_declared_job_type(job_type)?;
43        let handler_type = Self::validate_handler_job_type(&handler)?;
44        if declared != handler_type {
45            return Err(CatalogError::HandlerJobTypeMismatch {
46                declared: declared.as_str().to_owned(),
47                handler: handler_type.as_str().to_owned(),
48            });
49        }
50
51        let key = JobTypeName::new(job_type).map_err(|source| CatalogError::InvalidJobType {
52            job_type: job_type.to_owned(),
53            source,
54        })?;
55        if self.jobs.contains_key(&key) {
56            return Err(CatalogError::DuplicateJobType {
57                job_type: job_type.to_owned(),
58            });
59        }
60
61        self.jobs.insert(
62            key,
63            CatalogJob {
64                job_type: declared,
65                handler: Arc::new(handler),
66                definition_overrides: JobCatalogDefinitionOverrides::new(),
67                retry_delay_overrides: BTreeMap::new(),
68            },
69        );
70        Ok(self)
71    }
72
73    /// Registers a handler, panicking when validation fails.
74    #[must_use]
75    pub fn job<H>(self, job_type: &'static str, handler: H) -> Self
76    where
77        H: JobHandler + 'static,
78    {
79        self.try_job(job_type, handler).unwrap_or_else(|error| {
80            panic!("invalid job catalog registration for {job_type:?}: {error}");
81        })
82    }
83
84    /// Registers a handler with job-specific definition overrides.
85    ///
86    /// # Errors
87    /// Returns [`CatalogError`] when job types are blank, mismatched, or duplicated.
88    pub fn try_job_with_definition_overrides<H>(
89        self,
90        job_type: &'static str,
91        handler: H,
92        overrides: JobCatalogDefinitionOverrides,
93    ) -> Result<Self, CatalogError>
94    where
95        H: JobHandler + 'static,
96    {
97        self.try_job(job_type, handler)?
98            .try_definition_overrides(job_type, overrides)
99    }
100
101    /// Registers a handler with job-specific definition overrides, panicking
102    /// when validation fails.
103    #[must_use]
104    pub fn job_with_definition_overrides<H>(
105        self,
106        job_type: &'static str,
107        handler: H,
108        overrides: JobCatalogDefinitionOverrides,
109    ) -> Self
110    where
111        H: JobHandler + 'static,
112    {
113        self.try_job_with_definition_overrides(job_type, handler, overrides)
114            .unwrap_or_else(|error| {
115                panic!(
116                    "invalid job catalog registration with definition overrides for {job_type:?}: {error}"
117                );
118            })
119    }
120
121    /// Replaces the definition overrides for one registered catalog job.
122    ///
123    /// # Errors
124    /// Returns [`CatalogError::UnknownJobType`] when the job type is not registered.
125    pub fn try_definition_overrides(
126        mut self,
127        job_type: &str,
128        overrides: JobCatalogDefinitionOverrides,
129    ) -> Result<Self, CatalogError> {
130        let key = self.require_job_key(job_type)?;
131        overrides
132            .validate()
133            .map_err(|field| CatalogError::InvalidJobDefinitionValue {
134                job_type: job_type.to_owned(),
135                field,
136            })?;
137        self.jobs
138            .get_mut(&key)
139            .expect("job key validated")
140            .definition_overrides = overrides;
141        Ok(self)
142    }
143
144    /// Replaces the definition overrides for one registered catalog job,
145    /// panicking when validation fails.
146    #[must_use]
147    pub fn definition_overrides(
148        self,
149        job_type: &str,
150        overrides: JobCatalogDefinitionOverrides,
151    ) -> Self {
152        self.try_definition_overrides(job_type, overrides)
153            .unwrap_or_else(|error| {
154                panic!("invalid definition overrides for job type {job_type:?}: {error}");
155            })
156    }
157
158    /// Registers a retry-delay override for a catalog job type.
159    ///
160    /// # Errors
161    /// Returns [`CatalogError`] when the job type is unknown or override values are invalid.
162    pub fn try_retry_delay_override(
163        mut self,
164        job_type: &str,
165        failure_code: &'static str,
166        retry_delay_ms: i32,
167    ) -> Result<Self, CatalogError> {
168        let key = self.require_job_key(job_type)?;
169        Self::validate_failure_code(failure_code)?;
170        Self::validate_retry_delay(retry_delay_ms)?;
171        self.jobs
172            .get_mut(&key)
173            .expect("job key validated")
174            .retry_delay_overrides
175            .insert(failure_code, retry_delay_ms);
176        Ok(self)
177    }
178
179    /// Registers a retry-delay override, panicking when validation fails.
180    #[must_use]
181    pub fn retry_delay_override(
182        self,
183        job_type: &str,
184        failure_code: &'static str,
185        retry_delay_ms: i32,
186    ) -> Self {
187        self.try_retry_delay_override(job_type, failure_code, retry_delay_ms)
188            .unwrap_or_else(|error| {
189                panic!(
190                    "invalid retry delay override for job type {job_type:?}, failure code {failure_code:?}: {error}"
191                );
192            })
193    }
194
195    /// Registers a catalog-owned schedule after validating its shape and job type.
196    ///
197    /// Registered schedules are used by [`Self::sync_schedules`] and
198    /// [`Self::sync_schedules_exact`]. The referenced job must already be
199    /// registered on this builder and effectively enabled.
200    ///
201    /// # Errors
202    /// Returns [`CatalogError`] when the schedule spec is invalid, the job type
203    /// is unknown or disabled, or the schedule name is already registered.
204    pub fn try_schedule(mut self, spec: CatalogJobScheduleSpec<'_>) -> Result<Self, CatalogError> {
205        spec.validate_shape()
206            .map_err(|field| CatalogError::InvalidScheduleSpec {
207                name: spec.name.to_owned(),
208                field,
209            })?;
210        self.require_catalog_enabled_job_type(spec.job_type)?;
211
212        if self.schedules.iter().any(|stored| stored.name == spec.name) {
213            return Err(CatalogError::DuplicateScheduleName {
214                name: spec.name.to_owned(),
215            });
216        }
217
218        self.schedules
219            .push(StoredCatalogJobScheduleSpec::from(&spec));
220        Ok(self)
221    }
222
223    /// Registers a catalog-owned schedule, panicking when validation fails.
224    ///
225    /// Use [`Self::try_schedule`] when registration data is not static or should
226    /// be reported as a recoverable startup error.
227    #[must_use]
228    pub fn schedule(self, spec: CatalogJobScheduleSpec<'_>) -> Self {
229        self.try_schedule(spec).unwrap_or_else(|error| {
230            panic!("invalid job catalog schedule registration: {error}");
231        })
232    }
233
234    /// Converts the catalog into a runtime [`JobRegistry`].
235    ///
236    /// Disabled catalog jobs still register handlers so workers can process
237    /// already-queued work and dead-letter hooks.
238    #[must_use]
239    pub fn to_registry(&self) -> JobRegistry {
240        let mut registry = JobRegistry::new();
241        for entry in self.jobs.values() {
242            registry.register_boxed(Arc::clone(&entry.handler));
243            for (failure_code, retry_delay_ms) in &entry.retry_delay_overrides {
244                registry.register_retry_delay_override(
245                    entry.job_type,
246                    failure_code,
247                    *retry_delay_ms,
248                );
249            }
250        }
251        registry
252    }
253
254    /// Returns whether the catalog has a registered job type.
255    #[must_use]
256    pub fn contains(&self, job_type: JobType<'_>) -> bool {
257        self.jobs.contains_key(job_type.as_str())
258    }
259
260    /// Returns a catalog job type when it is registered.
261    ///
262    /// # Errors
263    /// Returns [`CatalogError::UnknownJobType`] when the name is not in the catalog.
264    pub fn require_job_type(&self, job_type: &str) -> Result<JobType<'static>, CatalogError> {
265        let key = self.require_job_key(job_type)?;
266        Ok(self.jobs.get(&key).expect("job key validated").job_type)
267    }
268
269    /// Returns a catalog job type when it is registered and catalog-enabled.
270    ///
271    /// This checks catalog configuration only. It does not read `job_definitions`;
272    /// operator-disabled database rows are enforced later by persistence APIs.
273    /// Job-specific definition overrides take precedence over the catalog
274    /// default enabled flag when present.
275    ///
276    /// # Errors
277    /// Returns [`CatalogError::UnknownJobType`] or [`CatalogError::DisabledJobType`].
278    pub fn require_catalog_enabled_job_type(
279        &self,
280        job_type: &str,
281    ) -> Result<JobType<'static>, CatalogError> {
282        let key = self.require_job_key(job_type)?;
283        let entry = self.jobs.get(&key).expect("job key validated");
284        if !self.effective_defaults(entry).is_enabled {
285            return Err(CatalogError::DisabledJobType {
286                job_type: entry.job_type.as_str().to_owned(),
287            });
288        }
289        Ok(entry.job_type)
290    }
291
292    pub(super) fn validate_defaults(&self) -> Result<(), CatalogError> {
293        self.defaults
294            .validate()
295            .map_err(|field| CatalogError::InvalidDefinitionValue { field })?;
296
297        for entry in self.jobs.values() {
298            self.effective_defaults(entry).validate().map_err(|field| {
299                CatalogError::InvalidJobDefinitionValue {
300                    job_type: entry.job_type.as_str().to_owned(),
301                    field,
302                }
303            })?;
304        }
305
306        Ok(())
307    }
308
309    pub(super) fn validate_exact_sync_scope(
310        &self,
311        scope: &JobCatalogSyncScope,
312    ) -> Result<(), CatalogError> {
313        if self.jobs.is_empty() {
314            return Err(CatalogError::EmptyExactSyncCatalog);
315        }
316
317        for entry in self.jobs.values() {
318            if !scope.contains(entry.job_type) {
319                return Err(CatalogError::JobTypeOutsideExactSyncScope {
320                    job_type: entry.job_type.as_str().to_owned(),
321                });
322            }
323        }
324
325        Ok(())
326    }
327
328    pub(super) fn materialize_definition(
329        &self,
330        entry: &CatalogJob,
331    ) -> JobDefinitionUpsert<'static> {
332        let defaults = self.effective_defaults(entry);
333        JobDefinitionUpsert {
334            job_type: entry.job_type,
335            version: defaults.version,
336            max_attempts: defaults.max_attempts,
337            default_timeout_seconds: defaults.default_timeout_seconds,
338            default_priority: defaults.default_priority,
339            is_enabled: defaults.is_enabled,
340        }
341    }
342
343    pub(super) fn effective_defaults(&self, entry: &CatalogJob) -> JobCatalogDefaults {
344        entry.definition_overrides.apply_to(self.defaults)
345    }
346
347    pub(super) fn require_job_key(&self, job_type: &str) -> Result<JobTypeName, CatalogError> {
348        let key = JobTypeName::new(job_type).map_err(|source| CatalogError::InvalidJobType {
349            job_type: job_type.to_owned(),
350            source,
351        })?;
352        if self.jobs.contains_key(&key) {
353            Ok(key)
354        } else {
355            Err(CatalogError::UnknownJobType {
356                job_type: job_type.to_owned(),
357            })
358        }
359    }
360
361    fn validate_declared_job_type(
362        job_type: &'static str,
363    ) -> Result<JobType<'static>, CatalogError> {
364        JobType::try_new(job_type).map_err(|source| CatalogError::InvalidJobType {
365            job_type: job_type.to_owned(),
366            source,
367        })
368    }
369
370    fn validate_handler_job_type<H: JobHandler + ?Sized>(
371        handler: &H,
372    ) -> Result<JobType<'static>, CatalogError> {
373        let handler_job_type = handler.job_type();
374        JobType::try_new(handler_job_type.as_str()).map_err(|source| {
375            CatalogError::InvalidHandlerJobType {
376                handler_job_type: handler_job_type.as_str().to_owned(),
377                source,
378            }
379        })
380    }
381
382    fn validate_failure_code(failure_code: &str) -> Result<(), CatalogError> {
383        if failure_code.trim().is_empty() {
384            Err(CatalogError::InvalidFailureCode)
385        } else {
386            Ok(())
387        }
388    }
389
390    fn validate_retry_delay(retry_delay_ms: i32) -> Result<(), CatalogError> {
391        if retry_delay_ms <= 0 {
392            Err(CatalogError::InvalidRetryDelay)
393        } else {
394            Ok(())
395        }
396    }
397}
398
399impl Default for JobCatalog {
400    fn default() -> Self {
401        Self::new()
402    }
403}
runledger_runtime/catalog/registration.rs

runledger_runtime/catalog/
registration.rs
