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-backed coordinator for split deployments (no local tick/worker loops).
130///
131/// Targets the `chronon-axum` API at `{base_url}/api/chronon/*`. Pair with
132/// [`DeploymentShape::RemoteClient`](crate::DeploymentShape::RemoteClient) on the worker host.
133pub struct RemoteCoordinatorClient {
134    base_url: String,
135    client: reqwest::Client,
136}
137
138impl RemoteCoordinatorClient {
139    /// `base_url` is trimmed of trailing slashes; timeout from `CHRONON_REMOTE_HTTP_TIMEOUT_MS` (default 3s).
140    pub fn new(base_url: impl Into<String>) -> Self {
141        Self {
142            base_url: trim_slash(&base_url.into()),
143            client: http_client(),
144        }
145    }
146
147    fn api_url(&self, path: &str) -> String {
148        format!("{}/api/chronon{}", self.base_url, path)
149    }
150
151    async fn parse_response<T: for<'de> Deserialize<'de>>(
152        resp: reqwest::Response,
153    ) -> Result<T> {
154        let status = resp.status();
155        let body: ApiResponse<T> = resp
156            .json()
157            .await
158            .map_err(|e| ChrononError::Internal(format!("remote decode: {e}")))?;
159        if !status.is_success() || !body.success {
160            return Err(ChrononError::Internal(
161                body.error.unwrap_or_else(|| format!("HTTP {status}")),
162            ));
163        }
164        body.data
165            .ok_or_else(|| ChrononError::Internal("remote empty data".into()))
166    }
167
168    /// POST `/jobs/upsert` with fields from `job`.
169    pub async fn upsert_job(&self, job: Job) -> Result<()> {
170        let req = UpsertJobRequest {
171            job_name: job.job_name,
172            script_name: job.script_name,
173            cron_expr: job.cron_expr,
174            timezone: job.timezone,
175            schedule_kind: job.schedule_kind.into(),
176            params: job.params_json,
177            enabled: job.enabled,
178            concurrency: job.concurrency,
179            timeout_ms: job.timeout_ms,
180            actor_json: Some(job.actor_json),
181            retry_policy: Some(job.retry_policy_json),
182            misfire_policy: Some(job.misfire_policy_json),
183        };
184        let _: JobResponse = Self::parse_response(
185            self.client
186                .post(self.api_url("/jobs/upsert"))
187                .json(&req)
188                .send()
189                .await
190                .map_err(|e| ChrononError::Internal(e.to_string()))?,
191        )
192        .await?;
193        Ok(())
194    }
195
196    /// GET `/jobs` and return non-sensitive job summaries.
197    pub async fn list_jobs(&self) -> Result<Vec<JobSummary>> {
198        Self::parse_response(
199            self.client
200                .get(self.api_url("/jobs"))
201                .send()
202                .await
203                .map_err(|e| ChrononError::Internal(e.to_string()))?,
204        )
205        .await
206    }
207
208    /// POST `/jobs/run_now` and return the enqueued `run_id`.
209    pub async fn run_now(&self, job_id: &str) -> Result<String> {
210        let req = JobActionRequest {
211            job_id: job_id.to_string(),
212            params: None,
213        };
214        let resp: String = Self::parse_response(
215            self.client
216                .post(self.api_url("/jobs/run_now"))
217                .json(&req)
218                .send()
219                .await
220                .map_err(|e| ChrononError::Internal(e.to_string()))?,
221        )
222        .await?;
223        Ok(resp)
224    }
225}
226
227/// Resolve remote base URL from `CHRONON_REMOTE_BASE_URL` when set and non-empty.
228pub fn resolve_remote_base_url() -> Option<String> {
229    if let Ok(u) = std::env::var("CHRONON_REMOTE_BASE_URL") {
230        let t = u.trim();
231        if !t.is_empty() {
232            return Some(trim_slash(t));
233        }
234    }
235    None
236}
chronon_runtime/remote_client.rs

chronon_runtime/
remote_client.rs
