runledger_runtime/catalog/
inputs.rs

1use chrono::{DateTime, Utc};
2use runledger_core::jobs::{JobStage, WorkflowStepEnqueueBuilder};
3use runledger_postgres::jobs::{JobEnqueue, JobScheduleUpsert};
4use serde_json::Value;
5use uuid::Uuid;
6
7use super::{CatalogError, JobCatalog};
8
9/// Input for building a [`JobEnqueue`] from a catalog-backed job type.
10#[derive(Debug, Clone)]
11pub struct CatalogJobEnqueueInput<'a> {
12    /// Catalog job type to enqueue.
13    pub job_type: &'a str,
14    /// Optional organization scope copied to the queued job.
15    pub organization_id: Option<Uuid>,
16    /// JSON payload stored with the queued job.
17    pub payload: &'a Value,
18    /// Optional queue priority override.
19    pub priority: Option<i32>,
20    /// Optional maximum-attempts override.
21    pub max_attempts: Option<i32>,
22    /// Optional execution timeout override, in seconds.
23    pub timeout_seconds: Option<i32>,
24    /// Optional future time before which the job should not run.
25    pub next_run_at: Option<DateTime<Utc>>,
26    /// Optional idempotency key for duplicate enqueue protection.
27    pub idempotency_key: Option<&'a str>,
28    /// Optional initial job stage.
29    pub stage: Option<JobStage>,
30}
31
32/// Input for building a one-off [`JobScheduleUpsert`] from a catalog-backed job type.
33///
34/// Prefer [`JobCatalog::schedule`](super::JobCatalog::schedule) plus
35/// [`JobCatalog::sync_schedules`](super::JobCatalog::sync_schedules) for
36/// catalog-owned schedules, or
37/// [`JobCatalog::sync_schedules_with`](super::JobCatalog::sync_schedules_with)
38/// for startup-provided catalog-owned specs. Use this lower-level input for
39/// migrations, admin tools, and schedules that should not be owned by catalog
40/// sync.
41#[derive(Debug, Clone)]
42pub struct CatalogJobScheduleInput<'a> {
43    /// Unique schedule name.
44    pub name: &'a str,
45    /// Catalog job type enqueued when the schedule fires.
46    pub job_type: &'a str,
47    /// Optional organization scope copied into jobs for a new schedule.
48    ///
49    /// Existing schedules preserve their stored organization scope on conflict.
50    pub organization_id: Option<Uuid>,
51    /// JSON payload template copied into scheduled jobs.
52    pub payload_template: &'a Value,
53    /// Cron expression used by the runtime scheduler.
54    pub cron_expr: &'a str,
55    /// Whether a new schedule should be active.
56    ///
57    /// Existing schedules preserve their stored active state on conflict.
58    pub is_active: bool,
59    /// Next UTC instant at which the schedule is due.
60    pub next_fire_at: DateTime<Utc>,
61    /// Maximum deterministic jitter, in seconds, applied to future fire times.
62    pub max_jitter_seconds: i32,
63}
64
65impl JobCatalog {
66    /// Builds a [`JobEnqueue`] after validating the job type is registered and enabled.
67    ///
68    /// This checks catalog configuration only. Operator-disabled database rows
69    /// are still enforced by `runledger-postgres` when the job is enqueued.
70    /// Job-specific definition overrides take precedence over the catalog
71    /// default enabled flag when present.
72    pub fn job_enqueue<'a>(
73        &self,
74        input: &CatalogJobEnqueueInput<'a>,
75    ) -> Result<JobEnqueue<'a>, CatalogError> {
76        let job_type = self.require_catalog_enabled_job_type(input.job_type)?;
77        Ok(JobEnqueue {
78            job_type,
79            organization_id: input.organization_id,
80            payload: input.payload,
81            priority: input.priority,
82            max_attempts: input.max_attempts,
83            timeout_seconds: input.timeout_seconds,
84            next_run_at: input.next_run_at,
85            idempotency_key: input.idempotency_key,
86            stage: input.stage,
87        })
88    }
89
90    /// Builds a one-off [`JobScheduleUpsert`] after validating the job type is
91    /// registered and enabled.
92    ///
93    /// This helper does not make the schedule catalog-owned. It is intended for
94    /// lower-level setup flows that will call
95    /// `runledger_postgres::jobs::upsert_job_schedule` directly.
96    ///
97    /// This checks catalog configuration only. Operator-disabled database rows
98    /// are still enforced by `runledger-postgres` when schedule-created jobs are
99    /// materialized.
100    /// Job-specific definition overrides take precedence over the catalog
101    /// default enabled flag when present.
102    pub fn job_schedule<'a>(
103        &self,
104        input: &CatalogJobScheduleInput<'a>,
105    ) -> Result<JobScheduleUpsert<'a>, CatalogError> {
106        let job_type = self.require_catalog_enabled_job_type(input.job_type)?;
107        Ok(JobScheduleUpsert {
108            name: input.name,
109            job_type,
110            organization_id: input.organization_id,
111            payload_template: input.payload_template,
112            cron_expr: input.cron_expr,
113            is_active: input.is_active,
114            next_fire_at: input.next_fire_at,
115            max_jitter_seconds: input.max_jitter_seconds,
116        })
117    }
118
119    /// Builds a workflow step after validating the job type is registered and enabled.
120    pub fn workflow_step<'a>(
121        &self,
122        step_key: &'a str,
123        job_type_name: &str,
124        payload: &'a Value,
125    ) -> Result<WorkflowStepEnqueueBuilder<'a>, CatalogError> {
126        let job_type = self.require_catalog_enabled_job_type(job_type_name)?;
127        WorkflowStepEnqueueBuilder::try_new(step_key, job_type.as_str(), payload)
128            .map_err(CatalogError::WorkflowBuild)
129    }
130}
runledger_runtime/catalog/inputs.rs

runledger_runtime/catalog/
inputs.rs
