chronon_runtime/
remote_client.rs

1//! Generic HTTP client for remote coordinator access.
2
3use std::time::Duration;
4
5use chronon_core::models::{Job, ScheduleKind};
6use chronon_core::{ChrononError, Result};
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9
10const DEFAULT_TIMEOUT_MS: u64 = 3000;
11
12fn trim_slash(s: &str) -> String {
13    s.trim_end_matches('/').to_string()
14}
15
16fn remote_timeout() -> Duration {
17    let ms = std::env::var("CHRONON_REMOTE_HTTP_TIMEOUT_MS")
18        .ok()
19        .and_then(|s| s.parse().ok())
20        .unwrap_or(DEFAULT_TIMEOUT_MS);
21    Duration::from_millis(ms.max(100))
22}
23
24fn http_client() -> reqwest::Client {
25    reqwest::Client::builder()
26        .timeout(remote_timeout())
27        .build()
28        .unwrap_or_else(|_| reqwest::Client::new())
29}
30
31#[derive(Debug, Serialize, Deserialize)]
32struct ApiResponse<T> {
33    success: bool,
34    data: Option<T>,
35    error: Option<String>,
36}
37
38#[derive(Debug, Deserialize, Serialize)]
39struct UpsertJobRequest {
40    job_name: String,
41    script_name: String,
42    cron_expr: Option<String>,
43    timezone: Option<String>,
44    #[serde(default)]
45    schedule_kind: ScheduleKindDto,
46    #[serde(default)]
47    params: Value,
48    #[serde(default = "default_true")]
49    enabled: bool,
50    #[serde(default = "default_concurrency")]
51    concurrency: i32,
52    pub timeout_ms: Option<i64>,
53    #[serde(default)]
54    actor_json: Option<Value>,
55    #[serde(default)]
56    retry_policy: Option<Value>,
57    #[serde(default)]
58    misfire_policy: Option<Value>,
59}
60
61fn default_true() -> bool {
62    true
63}
64
65fn default_concurrency() -> i32 {
66    1
67}
68
69#[derive(Debug, Default, Deserialize, Serialize, Clone, Copy)]
70#[serde(rename_all = "snake_case")]
71enum ScheduleKindDto {
72    #[default]
73    Cron,
74    RunOnce,
75    Manual,
76}
77
78impl From<ScheduleKind> for ScheduleKindDto {
79    fn from(k: ScheduleKind) -> Self {
80        match k {
81            ScheduleKind::Cron => ScheduleKindDto::Cron,
82            ScheduleKind::RunOnce => ScheduleKindDto::RunOnce,
83            ScheduleKind::Manual => ScheduleKindDto::Manual,
84        }
85    }
86}
87
88/// Non-sensitive job summary returned by remote `GET /jobs` (matches Axum `JobResponse`).
89#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
90pub struct JobSummary {
91    /// Stable job identifier (UUID).
92    pub job_id: String,
93    /// Human-readable name.
94    pub job_name: String,
95    /// Bound script name.
96    pub script_name: String,
97    /// Whether scheduling is active.
98    pub enabled: bool,
99    /// Schedule mode as snake_case string (`cron`, `run_once`, `manual`).
100    pub schedule_kind: String,
101    /// Cron string when applicable.
102    pub cron_expr: Option<String>,
103    /// Timezone used for cron.
104    pub timezone: Option<String>,
105    /// Next scheduled fire time (RFC3339) when known.
106    pub next_run_at: Option<String>,
107    /// Monotonic revision counter.
108    pub current_revision: i32,
109    /// Creation timestamp (RFC3339).
110    pub created_at: String,
111    /// Last upsert timestamp (RFC3339).
112    pub updated_at: String,
113}
114
115#[derive(Debug, Deserialize, Serialize)]
116struct JobResponse {
117    job_id: String,
118    job_name: String,
119    script_name: String,
120}
121
122#[derive(Debug, Serialize, Deserialize)]
123struct JobActionRequest {
124    job_id: String,
125    #[serde(default)]
126    params: Option<Value>,
127}
128
129/// HTTP client for Mode 3 — schedule/trigger jobs without local Chronon loops.
130///
131/// Targets the `chronon-axum` API at `{base_url}/api/chronon/*` (see `API_PREFIX`).
132/// Pair with a host that mounts `chronon_router` on a Mode 1 or Mode 2 coordinator process.
133///
134/// | Method | HTTP |
135/// |--------|------|
136/// | [`Self::upsert_job`] | `POST /jobs/upsert` |
137/// | [`Self::list_jobs`] | `GET /jobs` |
138/// | [`Self::run_now`] | `POST /jobs/run_now` |
139///
140/// Timeout: `CHRONON_REMOTE_HTTP_TIMEOUT_MS` (default 3000, minimum 100). Base URL helper:
141/// [`resolve_remote_base_url`]. Optional builder mark: [`crate::ChrononBuilder::remote_coordinator`]
142/// ([`DeploymentShape::RemoteClient`](crate::DeploymentShape::RemoteClient)) so you do not call
143/// [`crate::Chronon::run`] by mistake.
144///
145/// # Examples
146///
147/// ```no_run
148/// use chronon_core::{Job, ScheduleKind};
149/// use chronon_runtime::{resolve_remote_base_url, RemoteCoordinatorClient};
150///
151/// # async fn demo() -> chronon_core::Result<()> {
152/// let base = resolve_remote_base_url()
153///     .unwrap_or_else(|| "http://127.0.0.1:8080".into());
154/// let client = RemoteCoordinatorClient::new(base);
155///
156/// let mut job = Job::new("remote-job", "nightly_cleanup");
157/// job.schedule_kind = ScheduleKind::Manual;
158/// client.upsert_job(job.clone()).await?;
159/// let _run_id = client.run_now(&job.job_id).await?;
160/// let _jobs = client.list_jobs().await?;
161/// # Ok(())
162/// # }
163/// ```
164///
165/// Runnable API mount sketch: `cargo run -p uf-chronon --example axum_host --features mem,axum`.
166pub struct RemoteCoordinatorClient {
167    base_url: String,
168    client: reqwest::Client,
169}
170
171impl RemoteCoordinatorClient {
172    /// Create a client for `base_url` (trailing slashes stripped).
173    ///
174    /// Timeout from `CHRONON_REMOTE_HTTP_TIMEOUT_MS` (default 3s).
175    pub fn new(base_url: impl Into<String>) -> Self {
176        Self {
177            base_url: trim_slash(&base_url.into()),
178            client: http_client(),
179        }
180    }
181
182    fn api_url(&self, path: &str) -> String {
183        format!("{}/api/chronon{}", self.base_url, path)
184    }
185
186    async fn parse_response<T: for<'de> Deserialize<'de>>(
187        resp: reqwest::Response,
188    ) -> Result<T> {
189        let status = resp.status();
190        let body: ApiResponse<T> = resp
191            .json()
192            .await
193            .map_err(|e| ChrononError::Internal(format!("remote decode: {e}")))?;
194        if !status.is_success() || !body.success {
195            return Err(ChrononError::Internal(
196                body.error.unwrap_or_else(|| format!("HTTP {status}")),
197            ));
198        }
199        body.data
200            .ok_or_else(|| ChrononError::Internal("remote empty data".into()))
201    }
202
203    /// POST `/jobs/upsert` with fields from `job`.
204    pub async fn upsert_job(&self, job: Job) -> Result<()> {
205        let req = UpsertJobRequest {
206            job_name: job.job_name,
207            script_name: job.script_name,
208            cron_expr: job.cron_expr,
209            timezone: job.timezone,
210            schedule_kind: job.schedule_kind.into(),
211            params: job.params_json,
212            enabled: job.enabled,
213            concurrency: job.concurrency,
214            timeout_ms: job.timeout_ms,
215            actor_json: Some(job.actor_json),
216            retry_policy: Some(job.retry_policy_json),
217            misfire_policy: Some(job.misfire_policy_json),
218        };
219        let _: JobResponse = Self::parse_response(
220            self.client
221                .post(self.api_url("/jobs/upsert"))
222                .json(&req)
223                .send()
224                .await
225                .map_err(|e| ChrononError::Internal(e.to_string()))?,
226        )
227        .await?;
228        Ok(())
229    }
230
231    /// GET `/jobs` and return non-sensitive job summaries.
232    pub async fn list_jobs(&self) -> Result<Vec<JobSummary>> {
233        Self::parse_response(
234            self.client
235                .get(self.api_url("/jobs"))
236                .send()
237                .await
238                .map_err(|e| ChrononError::Internal(e.to_string()))?,
239        )
240        .await
241    }
242
243    /// POST `/jobs/run_now` and return the enqueued `run_id`.
244    pub async fn run_now(&self, job_id: &str) -> Result<String> {
245        let req = JobActionRequest {
246            job_id: job_id.to_string(),
247            params: None,
248        };
249        let resp: String = Self::parse_response(
250            self.client
251                .post(self.api_url("/jobs/run_now"))
252                .json(&req)
253                .send()
254                .await
255                .map_err(|e| ChrononError::Internal(e.to_string()))?,
256        )
257        .await?;
258        Ok(resp)
259    }
260}
261
262/// Resolve remote API base URL from `CHRONON_REMOTE_BASE_URL`.
263///
264/// Returns [`None`] when unset or whitespace-only. Non-empty values are trimmed and
265/// trailing `/` is stripped — pass the result to [`RemoteCoordinatorClient::new`] or
266/// [`crate::ChrononBuilder::remote_coordinator`].
267///
268/// # Examples
269///
270/// ```
271/// use chronon_runtime::{resolve_remote_base_url, RemoteCoordinatorClient};
272///
273/// let base = resolve_remote_base_url()
274///     .unwrap_or_else(|| "http://127.0.0.1:8080".into());
275/// let client = RemoteCoordinatorClient::new(base);
276/// let _ = client; // call upsert_job / run_now / list_jobs against a live API host
277/// ```
278pub fn resolve_remote_base_url() -> Option<String> {
279    if let Ok(u) = std::env::var("CHRONON_REMOTE_BASE_URL") {
280        let t = u.trim();
281        if !t.is_empty() {
282            return Some(trim_slash(t));
283        }
284    }
285    None
286}
chronon_runtime/remote_client.rs

chronon_runtime/
remote_client.rs
