tawasuyu/brahman

1//! ente-timer-compat: scheduler estilo cron + systemd .timer.
2//!
3//! Lee config en JSON desde `/etc/ente/timers.json` (override env
4//! `ENTE_TIMERS_FILE`):
5//!
6//! ```json
7//! [
8//!   {
9//!     "name": "daily-cleanup",
10//!     "schedule": "0 4 * * *",
11//!     "card": {
12//!       "id": "01KQ_TIMER_CLEANUP_0000000",
13//!       "label": "daily-cleanup-job",
14//!       "schema_version": 1,
15//!       "soma": {"namespaces": {}, "rlimits": {}, "cgroup": {"path": ""}},
16//!       "payload": {"Native": {"exec": "/usr/local/bin/cleanup", "argv": [], "envp": []}},
17//!       "supervision": "OneShot",
18//!       "provides": [], "requires": []
19//!     }
20//!   }
21//! ]
22//! ```
23//!
24//! Schedule: cron 5-fields `min hour dom mon dow` (DOM/DOW como en cron
25//! tradicional). `*` y `*/N` soportados, listas no.
26//!
27//! Cuando un timer dispara, se envía un `BusRequest::Invoke` al bus interno
28//! con la cap "TimerFire" + payload = serialized Card. Un Ente que provea
29//! esa cap (futuro: ente-zero internamente) hace el spawn.
30//!
31//! Para el demo: log "FIRE" cada vez que el schedule matchea, sin spawn real
32//! (requiere mover SpawnRequest al protocolo del bus, fuera de scope).
33
34use ente_bus::{BusClient, BusRequest, BusResponse};
35use ente_card::Capability;
36use serde::Deserialize;
37use std::time::{SystemTime, UNIX_EPOCH};
38use tokio::signal::unix::{signal, SignalKind};
39use tracing::{info, warn};
40use tracing_subscriber::EnvFilter;
41
42#[derive(Debug, Clone, Deserialize)]
43struct TimerConfig {
44    name: String,
45    /// Cron 5-field: `min hour dom mon dow`. `*`, `N`, `*/N` soportados.
46    schedule: String,
47    /// Card a disparar. Por ahora se loguea — futuro: SpawnRequest via bus.
48    #[serde(default)]
49    card: Option<serde_json::Value>,
50}
51
52#[derive(Debug)]
53struct Cron {
54    min: CronField,
55    hour: CronField,
56    dom: CronField,
57    mon: CronField,
58    dow: CronField,
59}
60
61#[derive(Debug)]
62enum CronField {
63    Any,
64    Exact(u32),
65    Step(u32),  // */N
66}
67
68impl CronField {
69    fn parse(s: &str) -> Option<Self> {
70        if s == "*" { return Some(CronField::Any); }
71        if let Some(n) = s.strip_prefix("*/") {
72            return n.parse().ok().map(CronField::Step);
73        }
74        s.parse().ok().map(CronField::Exact)
75    }
76    fn matches(&self, v: u32) -> bool {
77        match self {
78            CronField::Any => true,
79            CronField::Exact(n) => *n == v,
80            CronField::Step(n) if *n > 0 => v % n == 0,
81            _ => false,
82        }
83    }
84}
85
86impl Cron {
87    fn parse(s: &str) -> Option<Self> {
88        let parts: Vec<&str> = s.split_whitespace().collect();
89        if parts.len() != 5 { return None; }
90        Some(Self {
91            min:  CronField::parse(parts[0])?,
92            hour: CronField::parse(parts[1])?,
93            dom:  CronField::parse(parts[2])?,
94            mon:  CronField::parse(parts[3])?,
95            dow:  CronField::parse(parts[4])?,
96        })
97    }
98    fn matches(&self, t: &TimeBits) -> bool {
99        self.min.matches(t.min)
100            && self.hour.matches(t.hour)
101            && self.dom.matches(t.dom)
102            && self.mon.matches(t.mon)
103            && self.dow.matches(t.dow)
104    }
105}
106
107#[derive(Debug)]
108struct TimeBits {
109    min: u32, hour: u32, dom: u32, mon: u32, dow: u32,
110}
111
112/// Decompose epoch_secs en componentes UTC. Algoritmo simple (Howard Hinnant).
113fn time_bits_utc(epoch_secs: i64) -> TimeBits {
114    let secs_per_day = 86400i64;
115    let days_since_epoch = epoch_secs.div_euclid(secs_per_day);
116    let secs_in_day = epoch_secs.rem_euclid(secs_per_day);
117    let hour = (secs_in_day / 3600) as u32;
118    let min = ((secs_in_day % 3600) / 60) as u32;
119
120    // dow: 1970-01-01 fue jueves (4); cron usa 0-6 con 0=domingo.
121    let dow = ((days_since_epoch + 4).rem_euclid(7)) as u32;
122
123    // Conversión a y/m/d (Howard Hinnant Civil from days).
124    let z = days_since_epoch + 719_468;
125    let era = z.div_euclid(146_097);
126    let doe = (z - era * 146_097) as u64;
127    let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
128    let y = yoe as i64 + era * 400;
129    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
130    let mp = (5 * doy + 2) / 153;
131    let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
132    let m = (if mp < 10 { mp + 3 } else { mp - 9 }) as u32;
133    let _y = y + if m <= 2 { 1 } else { 0 };
134    TimeBits { min, hour, dom: d, mon: m, dow }
135}
136
137#[tokio::main(flavor = "current_thread")]
138async fn main() -> anyhow::Result<()> {
139    init_tracing();
140    info!("ente-timer-compat: arrancando");
141    announce_to_fractal().await;
142
143    let timers = load_timers();
144    info!(count = timers.len(), "timers cargados");
145    for t in &timers {
146        info!(name = %t.name, schedule = %t.schedule, "timer activo");
147    }
148
149    let parsed: Vec<(TimerConfig, Cron)> = timers.into_iter()
150        .filter_map(|t| {
151            let cron = Cron::parse(&t.schedule)?;
152            Some((t, cron))
153        })
154        .collect();
155
156    let mut term = signal(SignalKind::terminate())?;
157    let mut int_ = signal(SignalKind::interrupt())?;
158    let mut tick = tokio::time::interval(std::time::Duration::from_secs(60));
159    // Alinear al próximo minuto entero.
160    let now_ms = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64;
161    let to_next_min = 60_000 - (now_ms % 60_000);
162    tokio::time::sleep(std::time::Duration::from_millis(to_next_min)).await;
163    tick.tick().await; // descartar primer tick post-alignment
164
165    info!("scheduler activo (cron 5-field UTC)");
166    loop {
167        tokio::select! {
168            _ = tick.tick() => {
169                let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs() as i64;
170                let bits = time_bits_utc(now);
171                for (cfg, cron) in &parsed {
172                    if cron.matches(&bits) {
173                        fire(cfg).await;
174                    }
175                }
176            }
177            _ = term.recv() => { info!("SIGTERM"); return Ok(()); }
178            _ = int_.recv() => { info!("SIGINT"); return Ok(()); }
179        }
180    }
181}
182
183async fn fire(cfg: &TimerConfig) {
184    info!(name = %cfg.name, "TIMER FIRE");
185    if cfg.card.is_none() {
186        return;
187    }
188    // En el futuro: forwardear via bus a un proveedor que haga SpawnRequest.
189    // Por ahora log estructurado.
190    info!(name = %cfg.name, "card spawn requested (no-op por ahora)");
191}
192
193fn load_timers() -> Vec<TimerConfig> {
194    let path = std::env::var("ENTE_TIMERS_FILE")
195        .unwrap_or_else(|_| "/etc/ente/timers.json".into());
196    match std::fs::read_to_string(&path) {
197        Ok(content) => serde_json::from_str(&content).unwrap_or_else(|e| {
198            warn!(?e, path, "timers.json inválido — sin timers");
199            vec![]
200        }),
201        Err(_) => {
202            info!(path, "timers.json ausente — scheduler inactivo");
203            vec![]
204        }
205    }
206}
207
208async fn announce_to_fractal() {
209    if let Ok(mut client) = BusClient::from_env().await {
210        let req = BusRequest::Announce {
211            capabilities: vec![Capability::Endpoint {
212                interface: ente_card::InterfaceId([0xa8; 16]),
213                version: 1,
214            }],
215        };
216        match client.call(req).await {
217            Ok(BusResponse::Ok) => info!("Announce → bus interno OK"),
218            Ok(other) => warn!(?other, "Announce respuesta inesperada"),
219            Err(e) => warn!(?e, "Announce falló"),
220        }
221    }
222}
223
224fn init_tracing() {
225    let filter = EnvFilter::try_from_default_env()
226        .unwrap_or_else(|_| EnvFilter::new("ente_timer_compat=info"));
227    tracing_subscriber::fmt().with_env_filter(filter).with_target(true).init();
228}