1use std::collections::{BTreeSet, HashSet};
2
3use kbolt_types::{
4 AddScheduleRequest, KboltError, RemoveScheduleRequest, RemoveScheduleSelector,
5 ScheduleAddResponse, ScheduleDefinition, ScheduleInterval, ScheduleIntervalUnit,
6 ScheduleRemoveResponse, ScheduleScope, ScheduleTrigger, ScheduleWeekday,
7};
8
9use crate::lock::LockMode;
10use crate::schedule_backend::{current_schedule_backend, reconcile_schedule_backend};
11use crate::schedule_state_store::ScheduleRunStateStore;
12use crate::schedule_store::ScheduleCatalog;
13use crate::schedule_support::{schedule_id_number, schedule_id_sort_key};
14use crate::Result;
15
16use super::Engine;
17
18const MIN_SCHEDULE_INTERVAL_MINUTES: u32 = 5;
19
20impl Engine {
21 pub fn add_schedule(&self, req: AddScheduleRequest) -> Result<ScheduleAddResponse> {
22 self.add_schedule_with_backend_support_check(req, ensure_schedule_backend_supported)
23 }
24
25 fn add_schedule_with_backend_support_check(
26 &self,
27 req: AddScheduleRequest,
28 ensure_backend_supported: fn() -> Result<()>,
29 ) -> Result<ScheduleAddResponse> {
30 let _lock = self.acquire_operation_lock(LockMode::Exclusive)?;
31 ensure_backend_supported()?;
32 let trigger = normalize_schedule_trigger(req.trigger)?;
33 let scope = self.normalize_schedule_scope(req.scope, true)?;
34 let mut catalog = ScheduleCatalog::load(&self.config.config_dir)?;
35
36 if let Some(existing) = catalog
37 .schedules
38 .iter()
39 .find(|schedule| schedule.trigger == trigger && schedule.scope == scope)
40 {
41 return Err(KboltError::InvalidInput(format!(
42 "schedule already exists: {}",
43 existing.id
44 ))
45 .into());
46 }
47
48 let schedule_id = format!("s{}", catalog.next_id);
49 catalog.next_id = catalog.next_id.checked_add(1).ok_or_else(|| {
50 KboltError::InvalidInput("cannot create schedule: schedule ids exhausted".to_string())
51 })?;
52
53 let schedule = ScheduleDefinition {
54 id: schedule_id,
55 trigger,
56 scope,
57 };
58 catalog.schedules.push(schedule.clone());
59 catalog.save(&self.config.config_dir)?;
60 let backend = match reconcile_schedule_backend(
61 &self.config.config_dir,
62 &self.config.cache_dir,
63 &catalog.schedules,
64 ) {
65 Ok(backend) => backend,
66 Err(err) => {
67 return Err(KboltError::Internal(format!(
68 "schedule {} was saved, but backend reconcile failed: {err}",
69 schedule.id
70 ))
71 .into())
72 }
73 };
74
75 Ok(ScheduleAddResponse { schedule, backend })
76 }
77
78 pub fn list_schedules(&self) -> Result<Vec<ScheduleDefinition>> {
79 let mut schedules = ScheduleCatalog::load(&self.config.config_dir)?.schedules;
80 schedules.sort_by_key(|schedule| schedule_id_sort_key(&schedule.id));
81 Ok(schedules)
82 }
83
84 pub fn remove_schedule(&self, req: RemoveScheduleRequest) -> Result<ScheduleRemoveResponse> {
85 self.remove_schedule_with_backend_support_check(req, ensure_schedule_backend_supported)
86 }
87
88 fn remove_schedule_with_backend_support_check(
89 &self,
90 req: RemoveScheduleRequest,
91 ensure_backend_supported: fn() -> Result<()>,
92 ) -> Result<ScheduleRemoveResponse> {
93 let _lock = self.acquire_operation_lock(LockMode::Exclusive)?;
94 ensure_backend_supported()?;
95 let mut catalog = ScheduleCatalog::load(&self.config.config_dir)?;
96 let removed_ids = self.resolve_removed_schedule_ids(req.selector, &catalog.schedules)?;
97
98 if removed_ids.is_empty() {
99 reconcile_schedule_backend(
100 &self.config.config_dir,
101 &self.config.cache_dir,
102 &catalog.schedules,
103 )?;
104 return Ok(ScheduleRemoveResponse { removed_ids });
105 }
106
107 let removed = removed_ids.iter().cloned().collect::<HashSet<_>>();
108 catalog
109 .schedules
110 .retain(|schedule| !removed.contains(&schedule.id));
111 catalog.save(&self.config.config_dir)?;
112 for schedule_id in &removed_ids {
113 ScheduleRunStateStore::remove(&self.config.cache_dir, schedule_id)?;
114 }
115 if let Err(err) = reconcile_schedule_backend(
116 &self.config.config_dir,
117 &self.config.cache_dir,
118 &catalog.schedules,
119 ) {
120 return Err(KboltError::Internal(format!(
121 "removed schedules {}, but backend reconcile failed: {err}",
122 removed_ids.join(", ")
123 ))
124 .into());
125 }
126
127 Ok(ScheduleRemoveResponse { removed_ids })
128 }
129
130 fn normalize_schedule_scope(
131 &self,
132 scope: ScheduleScope,
133 validate_targets: bool,
134 ) -> Result<ScheduleScope> {
135 match scope {
136 ScheduleScope::All => Ok(ScheduleScope::All),
137 ScheduleScope::Space { space } => {
138 let normalized_space = normalize_scope_name("space", &space)?;
139 if validate_targets {
140 let resolved = self.storage.get_space(&normalized_space)?;
141 return Ok(ScheduleScope::Space {
142 space: resolved.name,
143 });
144 }
145
146 Ok(ScheduleScope::Space {
147 space: normalized_space,
148 })
149 }
150 ScheduleScope::Collections { space, collections } => {
151 let normalized_space = normalize_scope_name("space", &space)?;
152 let normalized_collections = normalize_collection_names(collections)?;
153
154 if validate_targets {
155 let resolved = self.storage.get_space(&normalized_space)?;
156 for collection in &normalized_collections {
157 self.storage.get_collection(resolved.id, collection)?;
158 }
159 return Ok(ScheduleScope::Collections {
160 space: resolved.name,
161 collections: normalized_collections,
162 });
163 }
164
165 Ok(ScheduleScope::Collections {
166 space: normalized_space,
167 collections: normalized_collections,
168 })
169 }
170 }
171 }
172
173 fn resolve_removed_schedule_ids(
174 &self,
175 selector: RemoveScheduleSelector,
176 schedules: &[ScheduleDefinition],
177 ) -> Result<Vec<String>> {
178 match selector {
179 RemoveScheduleSelector::All => Ok(schedules
180 .iter()
181 .map(|schedule| schedule.id.clone())
182 .collect()),
183 RemoveScheduleSelector::Id { id } => {
184 let normalized_id = normalize_schedule_id(&id)?;
185 if schedules
186 .iter()
187 .any(|schedule| schedule.id == normalized_id)
188 {
189 return Ok(vec![normalized_id]);
190 }
191
192 Err(KboltError::ScheduleNotFound { id: normalized_id }.into())
193 }
194 RemoveScheduleSelector::Scope { scope } => {
195 let normalized_scope = self.normalize_schedule_scope(scope, false)?;
196 let mut matches = schedules
197 .iter()
198 .filter(|schedule| schedule.scope == normalized_scope)
199 .map(|schedule| schedule.id.clone())
200 .collect::<Vec<_>>();
201 matches.sort_by_key(|id| schedule_id_sort_key(id));
202
203 match matches.len() {
204 0 => Err(KboltError::ScheduleScopeNotFound.into()),
205 1 => Ok(matches),
206 _ => Err(KboltError::ScheduleScopeAmbiguous { ids: matches }.into()),
207 }
208 }
209 }
210 }
211}
212
213pub(super) fn load_schedule_definition(
214 config_dir: &std::path::Path,
215 id: &str,
216) -> Result<ScheduleDefinition> {
217 let normalized_id = normalize_schedule_id(id)?;
218 let catalog = ScheduleCatalog::load(config_dir)?;
219 catalog
220 .schedules
221 .into_iter()
222 .find(|schedule| schedule.id == normalized_id)
223 .ok_or_else(|| KboltError::ScheduleNotFound { id: normalized_id }.into())
224}
225
226fn normalize_schedule_trigger(trigger: ScheduleTrigger) -> Result<ScheduleTrigger> {
227 match trigger {
228 ScheduleTrigger::Every { interval } => Ok(ScheduleTrigger::Every {
229 interval: normalize_schedule_interval(interval)?,
230 }),
231 ScheduleTrigger::Daily { time } => Ok(ScheduleTrigger::Daily {
232 time: normalize_schedule_time(&time)?,
233 }),
234 ScheduleTrigger::Weekly { weekdays, time } => Ok(ScheduleTrigger::Weekly {
235 weekdays: normalize_schedule_weekdays(weekdays)?,
236 time: normalize_schedule_time(&time)?,
237 }),
238 }
239}
240
241fn normalize_schedule_interval(interval: ScheduleInterval) -> Result<ScheduleInterval> {
242 if interval.value == 0 {
243 return Err(KboltError::InvalidScheduleInterval {
244 reason: "must be greater than zero".to_string(),
245 }
246 .into());
247 }
248
249 match interval.unit {
250 ScheduleIntervalUnit::Minutes if interval.value < MIN_SCHEDULE_INTERVAL_MINUTES => {
251 Err(KboltError::InvalidScheduleInterval {
252 reason: format!("must be at least {MIN_SCHEDULE_INTERVAL_MINUTES} minutes"),
253 }
254 .into())
255 }
256 ScheduleIntervalUnit::Minutes | ScheduleIntervalUnit::Hours => Ok(interval),
257 }
258}
259
260fn normalize_schedule_weekdays(weekdays: Vec<ScheduleWeekday>) -> Result<Vec<ScheduleWeekday>> {
261 let normalized = weekdays.into_iter().collect::<BTreeSet<_>>();
262 if normalized.is_empty() {
263 return Err(KboltError::InvalidInput(
264 "weekly schedules require at least one weekday".to_string(),
265 )
266 .into());
267 }
268
269 Ok(normalized.into_iter().collect())
270}
271
272fn normalize_schedule_time(input: &str) -> Result<String> {
273 let trimmed = input.trim();
274 if trimmed.is_empty() {
275 return Err(KboltError::InvalidInput("schedule time must not be empty".to_string()).into());
276 }
277
278 let collapsed = trimmed
279 .chars()
280 .filter(|ch| !ch.is_whitespace())
281 .collect::<String>()
282 .to_ascii_lowercase();
283
284 let (time_part, meridiem) = if let Some(time) = collapsed.strip_suffix("am") {
285 (time, Some("am"))
286 } else if let Some(time) = collapsed.strip_suffix("pm") {
287 (time, Some("pm"))
288 } else {
289 (collapsed.as_str(), None)
290 };
291
292 if time_part.is_empty() {
293 return Err(invalid_schedule_time(input).into());
294 }
295
296 let (mut hour, minute) = if let Some((hour_part, minute_part)) = time_part.split_once(':') {
297 if minute_part.contains(':') {
298 return Err(invalid_schedule_time(input).into());
299 }
300 (
301 parse_time_component(hour_part, input)?,
302 parse_time_component(minute_part, input)?,
303 )
304 } else {
305 if meridiem.is_none() {
306 return Err(invalid_schedule_time(input).into());
307 }
308 (parse_time_component(time_part, input)?, 0)
309 };
310
311 if minute > 59 {
312 return Err(invalid_schedule_time(input).into());
313 }
314
315 match meridiem {
316 Some("am") => {
317 if hour == 0 || hour > 12 {
318 return Err(invalid_schedule_time(input).into());
319 }
320 if hour == 12 {
321 hour = 0;
322 }
323 }
324 Some("pm") => {
325 if hour == 0 || hour > 12 {
326 return Err(invalid_schedule_time(input).into());
327 }
328 if hour != 12 {
329 hour += 12;
330 }
331 }
332 None => {
333 if hour > 23 {
334 return Err(invalid_schedule_time(input).into());
335 }
336 }
337 Some(_) => unreachable!("only am/pm meridiems are supported"),
338 }
339
340 Ok(format!("{hour:02}:{minute:02}"))
341}
342
343fn parse_time_component(component: &str, input: &str) -> Result<u32> {
344 if component.is_empty() {
345 return Err(invalid_schedule_time(input).into());
346 }
347
348 component
349 .parse::<u32>()
350 .map_err(|_| invalid_schedule_time(input).into())
351}
352
353fn invalid_schedule_time(input: &str) -> KboltError {
354 KboltError::InvalidInput(format!(
355 "invalid schedule time '{input}': use HH:MM, 3pm, or 3:00pm"
356 ))
357}
358
359fn normalize_scope_name(label: &str, name: &str) -> Result<String> {
360 let trimmed = name.trim();
361 if trimmed.is_empty() {
362 return Err(KboltError::InvalidInput(format!("{label} name must not be empty")).into());
363 }
364 Ok(trimmed.to_string())
365}
366
367fn normalize_collection_names(collections: Vec<String>) -> Result<Vec<String>> {
368 let mut normalized = BTreeSet::new();
369 for collection in collections {
370 let trimmed = collection.trim();
371 if trimmed.is_empty() {
372 return Err(
373 KboltError::InvalidInput("collection names must not be empty".to_string()).into(),
374 );
375 }
376 normalized.insert(trimmed.to_string());
377 }
378
379 if normalized.is_empty() {
380 return Err(KboltError::InvalidInput(
381 "collection scope must include at least one collection".to_string(),
382 )
383 .into());
384 }
385
386 Ok(normalized.into_iter().collect())
387}
388
389fn normalize_schedule_id(id: &str) -> Result<String> {
390 let normalized = id.trim().to_ascii_lowercase();
391 if normalized.is_empty() {
392 return Err(KboltError::InvalidInput("schedule id must not be empty".to_string()).into());
393 }
394
395 if schedule_id_number(&normalized).is_none() {
396 return Err(KboltError::InvalidInput(format!("invalid schedule id: {id}")).into());
397 }
398
399 Ok(normalized)
400}
401
402fn ensure_schedule_backend_supported() -> Result<()> {
403 current_schedule_backend().map(|_| ())
404}
405
406#[cfg(test)]
407mod tests {
408 use fs2::FileExt;
409 use std::fs::OpenOptions;
410 use std::mem;
411
412 use tempfile::tempdir;
413
414 use super::Engine;
415 use crate::config::{ChunkingConfig, Config, RankingConfig, ReapingConfig};
416 use crate::schedule_state_store::ScheduleRunStateStore;
417 use crate::schedule_store::ScheduleCatalog;
418 use crate::storage::Storage;
419 use kbolt_types::{
420 AddScheduleRequest, KboltError, RemoveScheduleRequest, RemoveScheduleSelector,
421 ScheduleRunResult, ScheduleRunState, ScheduleScope, ScheduleTrigger,
422 };
423
424 #[test]
425 fn add_schedule_does_not_persist_when_backend_support_check_fails() {
426 let engine = test_engine();
427
428 let err = engine
429 .add_schedule_with_backend_support_check(
430 AddScheduleRequest {
431 trigger: ScheduleTrigger::Daily {
432 time: "09:00".to_string(),
433 },
434 scope: ScheduleScope::All,
435 },
436 unsupported_schedule_backend,
437 )
438 .expect_err("unsupported backend should fail");
439
440 match KboltError::from(err) {
441 KboltError::InvalidInput(message) => {
442 assert!(
443 message.contains("schedule is not supported on this platform"),
444 "unexpected message: {message}"
445 );
446 }
447 other => panic!("unexpected error: {other}"),
448 }
449
450 let catalog =
451 ScheduleCatalog::load(&engine.config.config_dir).expect("load unchanged schedule file");
452 assert_eq!(catalog, ScheduleCatalog::default());
453 }
454
455 #[test]
456 fn remove_schedule_does_not_mutate_when_backend_support_check_fails() {
457 let engine = test_engine();
458 let added = engine
459 .add_schedule(AddScheduleRequest {
460 trigger: ScheduleTrigger::Daily {
461 time: "09:00".to_string(),
462 },
463 scope: ScheduleScope::All,
464 })
465 .expect("add schedule");
466 let saved_state = ScheduleRunState {
467 last_started: Some("2026-03-07T12:00:00Z".to_string()),
468 last_finished: Some("2026-03-07T12:00:09Z".to_string()),
469 last_result: Some(ScheduleRunResult::Success),
470 last_error: None,
471 };
472 ScheduleRunStateStore::save(&engine.config.cache_dir, &added.schedule.id, &saved_state)
473 .expect("save run state");
474
475 let err = engine
476 .remove_schedule_with_backend_support_check(
477 RemoveScheduleRequest {
478 selector: RemoveScheduleSelector::Id {
479 id: added.schedule.id.clone(),
480 },
481 },
482 unsupported_schedule_backend,
483 )
484 .expect_err("unsupported backend should fail");
485
486 match KboltError::from(err) {
487 KboltError::InvalidInput(message) => {
488 assert!(
489 message.contains("schedule is not supported on this platform"),
490 "unexpected message: {message}"
491 );
492 }
493 other => panic!("unexpected error: {other}"),
494 }
495
496 let catalog =
497 ScheduleCatalog::load(&engine.config.config_dir).expect("load unchanged schedule file");
498 assert_eq!(catalog.schedules, vec![added.schedule.clone()]);
499 let loaded_state =
500 ScheduleRunStateStore::load(&engine.config.cache_dir, &added.schedule.id)
501 .expect("load preserved run state");
502 assert_eq!(loaded_state, saved_state);
503 }
504
505 #[test]
506 fn list_schedules_succeeds_while_global_lock_is_held() {
507 let engine = test_engine();
508 let added = engine
509 .add_schedule(AddScheduleRequest {
510 trigger: ScheduleTrigger::Daily {
511 time: "09:00".to_string(),
512 },
513 scope: ScheduleScope::All,
514 })
515 .expect("add schedule");
516
517 let lock_path = engine.config.cache_dir.join("kbolt.lock");
518 std::fs::create_dir_all(&engine.config.cache_dir).expect("create cache dir");
519 let holder = OpenOptions::new()
520 .read(true)
521 .write(true)
522 .create(true)
523 .truncate(false)
524 .open(&lock_path)
525 .expect("open lock file");
526 FileExt::try_lock_exclusive(&holder).expect("acquire global lock");
527
528 let schedules = engine.list_schedules().expect("list schedules");
529 assert_eq!(schedules, vec![added.schedule]);
530 }
531
532 fn unsupported_schedule_backend() -> crate::Result<()> {
533 Err(
534 KboltError::InvalidInput("schedule is not supported on this platform".to_string())
535 .into(),
536 )
537 }
538
539 fn test_engine() -> Engine {
540 let root = tempdir().expect("create temp root");
541 let root_path = root.path().to_path_buf();
542 mem::forget(root);
543 let config_dir = root_path.join("config");
544 let cache_dir = root_path.join("cache");
545 let storage = Storage::new(&cache_dir).expect("create storage");
546 let config = Config {
547 config_dir,
548 cache_dir,
549 default_space: None,
550 providers: std::collections::HashMap::new(),
551 roles: crate::config::RoleBindingsConfig::default(),
552 reaping: ReapingConfig { days: 7 },
553 chunking: ChunkingConfig::default(),
554 ranking: RankingConfig::default(),
555 };
556 Engine::from_parts(storage, config)
557 }
558}