ironflow_runtime/trigger/polling/mod.rs
1//! Polling trigger for external data sources.
2//!
3//! [`PollingTrigger`] periodically executes a [`PollingProbe`] and emits a
4//! [`TriggerEvent`] when the probe detects new data.
5//! Optional deduplication prevents re-triggering when the probe result has
6//! not changed since the last poll.
7//!
8//! # Built-in probes
9//!
10//! - [`HttpProbe`](http::HttpProbe) -- polls an HTTP endpoint, triggers when the
11//! response body changes (behind the `trigger-polling-http` feature flag).
12//! - [`SqlProbe`](sql::SqlProbe) -- executes a SQL query, triggers when the
13//! result set is non-empty (behind the `trigger-polling-sql` feature flag).
14//!
15//! # Custom probes
16//!
17//! Implement [`PollingProbe`] to add your own data source:
18//!
19//! ```no_run
20//! use ironflow_runtime::trigger::polling::{PollingProbe, ProbeFuture, ProbeResult, ProbeError};
21//!
22//! struct MyProbe;
23//!
24//! impl PollingProbe for MyProbe {
25//! fn name(&self) -> &str { "my-probe" }
26//!
27//! fn poll(&self) -> ProbeFuture<'_> {
28//! Box::pin(async {
29//! Ok(Some(ProbeResult::new(serde_json::json!({"rows": 42}))))
30//! })
31//! }
32//! }
33//! ```
34//!
35//! # Examples
36//!
37//! ```no_run
38//! use std::time::Duration;
39//! use ironflow_runtime::trigger::polling::{
40//! PollingTrigger, PollingTriggerConfig, PollingProbe, ProbeFuture, ProbeResult,
41//! ProbeError,
42//! };
43//!
44//! struct StubProbe;
45//! impl PollingProbe for StubProbe {
46//! fn name(&self) -> &str { "stub" }
47//! fn poll(&self) -> ProbeFuture<'_> {
48//! Box::pin(async { Ok(Some(ProbeResult::new(serde_json::json!({"ok": true})))) })
49//! }
50//! }
51//!
52//! let trigger = PollingTrigger::new(PollingTriggerConfig {
53//! interval: Duration::from_secs(30),
54//! probe: Box::new(StubProbe),
55//! workflow_name: "ingest".to_string(),
56//! dedup: true,
57//! });
58//! ```
59
60#[cfg(feature = "trigger-polling-http")]
61pub mod http;
62#[cfg(feature = "trigger-polling-sql")]
63pub mod sql;
64
65use std::future::Future;
66use std::pin::Pin;
67use std::time::Duration;
68
69use serde_json::Value;
70use sha2::{Digest, Sha256};
71use tokio::time::interval;
72use tokio_util::sync::CancellationToken;
73use tracing::{info, warn};
74
75use ironflow_store::entities::TriggerKind;
76
77use super::{Trigger, TriggerEvent, TriggerFuture, TriggerSink};
78
79/// Future returned by [`PollingProbe::poll`].
80///
81/// Returns `Ok(Some(result))` when data is found, `Ok(None)` when the
82/// source has nothing to report, or `Err` on failure.
83pub type ProbeFuture<'a> =
84 Pin<Box<dyn Future<Output = Result<Option<ProbeResult>, ProbeError>> + Send + 'a>>;
85
86/// A probe that checks an external data source for new data.
87///
88/// Implementations are called periodically by [`PollingTrigger`]. The probe
89/// returns `Some(ProbeResult)` when data is found, or `None` when the
90/// source has nothing to report (e.g. a SQL query returned zero rows).
91///
92/// # Examples
93///
94/// ```no_run
95/// use ironflow_runtime::trigger::polling::{PollingProbe, ProbeFuture, ProbeResult, ProbeError};
96///
97/// struct AlwaysNewProbe;
98///
99/// impl PollingProbe for AlwaysNewProbe {
100/// fn name(&self) -> &str { "always-new" }
101///
102/// fn poll(&self) -> ProbeFuture<'_> {
103/// Box::pin(async {
104/// Ok(Some(ProbeResult::new(serde_json::json!({"data": "fresh"}))))
105/// })
106/// }
107/// }
108/// ```
109pub trait PollingProbe: Send + Sync {
110 /// Human-readable name for logging and metrics.
111 fn name(&self) -> &str;
112
113 /// Execute the probe and return the result.
114 ///
115 /// # Errors
116 ///
117 /// Returns [`ProbeError`] if the probe cannot reach the data source
118 /// or encounters an unrecoverable error.
119 fn poll(&self) -> ProbeFuture<'_>;
120}
121
122/// The result of a successful probe execution.
123///
124/// Contains the data payload and a SHA-256 content hash used for
125/// deduplication.
126///
127/// # Examples
128///
129/// ```
130/// use ironflow_runtime::trigger::polling::ProbeResult;
131/// use serde_json::json;
132///
133/// let result = ProbeResult::new(json!({"rows": 5}));
134/// assert!(!result.content_hash().is_empty());
135/// assert_eq!(result.data()["rows"], 5);
136/// ```
137#[derive(Debug, Clone)]
138pub struct ProbeResult {
139 data: Value,
140 content_hash: String,
141}
142
143impl ProbeResult {
144 /// Create a new probe result from a JSON payload.
145 ///
146 /// The content hash is computed as SHA-256 of the canonical JSON
147 /// representation.
148 ///
149 /// # Examples
150 ///
151 /// ```
152 /// use ironflow_runtime::trigger::polling::ProbeResult;
153 /// use serde_json::json;
154 ///
155 /// let r1 = ProbeResult::new(json!({"a": 1}));
156 /// let r2 = ProbeResult::new(json!({"a": 1}));
157 /// assert_eq!(r1.content_hash(), r2.content_hash());
158 ///
159 /// let r3 = ProbeResult::new(json!({"a": 2}));
160 /// assert_ne!(r1.content_hash(), r3.content_hash());
161 /// ```
162 pub fn new(data: Value) -> Self {
163 let json_bytes = serde_json::to_vec(&data).unwrap_or_default();
164 let hash = Sha256::digest(&json_bytes);
165 let content_hash = hex::encode(hash);
166 Self { data, content_hash }
167 }
168
169 /// Create a new probe result with a pre-computed content hash.
170 ///
171 /// Use this when the hash source differs from the JSON payload
172 /// (e.g. the raw HTTP response body before parsing).
173 ///
174 /// # Examples
175 ///
176 /// ```
177 /// use ironflow_runtime::trigger::polling::ProbeResult;
178 /// use serde_json::json;
179 ///
180 /// let result = ProbeResult::with_hash(json!({"body": "..."}), "abc123".to_string());
181 /// assert_eq!(result.content_hash(), "abc123");
182 /// ```
183 pub fn with_hash(data: Value, content_hash: String) -> Self {
184 Self { data, content_hash }
185 }
186
187 /// The data payload.
188 ///
189 /// # Examples
190 ///
191 /// ```
192 /// use ironflow_runtime::trigger::polling::ProbeResult;
193 /// use serde_json::json;
194 ///
195 /// let result = ProbeResult::new(json!({"count": 3}));
196 /// assert_eq!(result.data()["count"], 3);
197 /// ```
198 pub fn data(&self) -> &Value {
199 &self.data
200 }
201
202 /// The SHA-256 content hash.
203 ///
204 /// # Examples
205 ///
206 /// ```
207 /// use ironflow_runtime::trigger::polling::ProbeResult;
208 /// use serde_json::json;
209 ///
210 /// let result = ProbeResult::new(json!({"key": "val"}));
211 /// assert_eq!(result.content_hash().len(), 64);
212 /// ```
213 pub fn content_hash(&self) -> &str {
214 &self.content_hash
215 }
216}
217
218/// Error type for probe operations.
219///
220/// # Examples
221///
222/// ```
223/// use ironflow_runtime::trigger::polling::ProbeError;
224///
225/// let err = ProbeError::Failed("connection refused".to_string());
226/// assert!(err.to_string().contains("connection refused"));
227/// ```
228#[derive(Debug, thiserror::Error)]
229pub enum ProbeError {
230 /// The probe encountered an unrecoverable error.
231 #[error("probe failed: {0}")]
232 Failed(String),
233}
234
235/// Configuration for a [`PollingTrigger`].
236///
237/// # Examples
238///
239/// ```no_run
240/// use std::time::Duration;
241/// use ironflow_runtime::trigger::polling::{
242/// PollingTriggerConfig, PollingProbe, ProbeFuture, ProbeResult,
243/// };
244///
245/// struct Stub;
246/// impl PollingProbe for Stub {
247/// fn name(&self) -> &str { "stub" }
248/// fn poll(&self) -> ProbeFuture<'_> {
249/// Box::pin(async { Ok(Some(ProbeResult::new(serde_json::json!(null)))) })
250/// }
251/// }
252///
253/// let config = PollingTriggerConfig {
254/// interval: Duration::from_secs(60),
255/// probe: Box::new(Stub),
256/// workflow_name: "my-workflow".to_string(),
257/// dedup: true,
258/// };
259/// ```
260pub struct PollingTriggerConfig {
261 /// How often to poll.
262 pub interval: Duration,
263 /// The probe to execute.
264 pub probe: Box<dyn PollingProbe>,
265 /// The workflow to trigger when the probe detects new data.
266 pub workflow_name: String,
267 /// When `true`, skip triggering if the probe result hash matches the
268 /// previous one.
269 pub dedup: bool,
270}
271
272/// Metric name constants for polling triggers.
273#[cfg(feature = "prometheus")]
274mod metric_names {
275 /// Counter incremented on every poll attempt.
276 pub const POLLING_TRIGGER_TOTAL: &str = "ironflow_polling_trigger_total";
277}
278
279/// A trigger that periodically polls an external source.
280///
281/// See the [module-level documentation](self) for usage examples.
282///
283/// # Examples
284///
285/// ```no_run
286/// use std::time::Duration;
287/// use ironflow_runtime::trigger::polling::{
288/// PollingTrigger, PollingTriggerConfig, PollingProbe, ProbeFuture, ProbeResult,
289/// };
290///
291/// struct Stub;
292/// impl PollingProbe for Stub {
293/// fn name(&self) -> &str { "stub" }
294/// fn poll(&self) -> ProbeFuture<'_> {
295/// Box::pin(async { Ok(Some(ProbeResult::new(serde_json::json!(null)))) })
296/// }
297/// }
298///
299/// let trigger = PollingTrigger::new(PollingTriggerConfig {
300/// interval: Duration::from_secs(30),
301/// probe: Box::new(Stub),
302/// workflow_name: "ingest".to_string(),
303/// dedup: false,
304/// });
305/// ```
306pub struct PollingTrigger {
307 config: PollingTriggerConfig,
308}
309
310impl PollingTrigger {
311 /// Create a new polling trigger.
312 ///
313 /// # Examples
314 ///
315 /// ```no_run
316 /// use std::time::Duration;
317 /// use ironflow_runtime::trigger::polling::{
318 /// PollingTrigger, PollingTriggerConfig, PollingProbe, ProbeFuture, ProbeResult,
319 /// };
320 ///
321 /// struct Stub;
322 /// impl PollingProbe for Stub {
323 /// fn name(&self) -> &str { "stub" }
324 /// fn poll(&self) -> ProbeFuture<'_> {
325 /// Box::pin(async { Ok(Some(ProbeResult::new(serde_json::json!(null)))) })
326 /// }
327 /// }
328 ///
329 /// let trigger = PollingTrigger::new(PollingTriggerConfig {
330 /// interval: Duration::from_secs(10),
331 /// probe: Box::new(Stub),
332 /// workflow_name: "check".to_string(),
333 /// dedup: true,
334 /// });
335 /// ```
336 pub fn new(config: PollingTriggerConfig) -> Self {
337 Self { config }
338 }
339
340 /// Record a Prometheus metric for a poll outcome.
341 #[cfg(feature = "prometheus")]
342 fn record_metric(probe_name: &str, outcome: &str) {
343 use metrics::counter;
344
345 counter!(
346 metric_names::POLLING_TRIGGER_TOTAL,
347 "probe" => probe_name.to_string(),
348 "outcome" => outcome.to_string()
349 )
350 .increment(1);
351 }
352}
353
354impl Trigger for PollingTrigger {
355 fn name(&self) -> &str {
356 "polling-trigger"
357 }
358
359 fn start<'a>(&'a self, sink: TriggerSink, token: &'a CancellationToken) -> TriggerFuture<'a> {
360 Box::pin(async move {
361 let mut ticker = interval(self.config.interval);
362 let mut last_hash: Option<String> = None;
363 let probe_name = self.config.probe.name().to_string();
364
365 info!(
366 probe = %probe_name,
367 interval_secs = self.config.interval.as_secs(),
368 workflow = %self.config.workflow_name,
369 dedup = self.config.dedup,
370 "polling trigger started"
371 );
372
373 loop {
374 tokio::select! {
375 _ = token.cancelled() => {
376 info!(probe = %probe_name, "polling trigger shutting down");
377 return Ok(());
378 }
379 _ = ticker.tick() => {
380 match self.config.probe.poll().await {
381 Ok(Some(result)) => {
382 let should_trigger = if self.config.dedup {
383 match &last_hash {
384 Some(prev) => prev != result.content_hash(),
385 None => true,
386 }
387 } else {
388 true
389 };
390
391 if should_trigger {
392 let event = TriggerEvent {
393 workflow_name: self.config.workflow_name.clone(),
394 payload: result.data().clone(),
395 trigger_kind: TriggerKind::Polling {
396 probe: probe_name.clone(),
397 },
398 };
399
400 if let Err(e) = sink.send(event).await {
401 warn!(
402 probe = %probe_name,
403 error = %e,
404 "failed to emit polling trigger event"
405 );
406 return Err(e);
407 }
408
409 info!(
410 probe = %probe_name,
411 workflow = %self.config.workflow_name,
412 "polling trigger fired"
413 );
414 #[cfg(feature = "prometheus")]
415 Self::record_metric(&probe_name, "triggered");
416
417 last_hash = Some(result.content_hash().to_string());
418 } else {
419 info!(
420 probe = %probe_name,
421 "poll result unchanged, skipping"
422 );
423 #[cfg(feature = "prometheus")]
424 Self::record_metric(&probe_name, "unchanged");
425 }
426 }
427 Ok(None) => {
428 info!(
429 probe = %probe_name,
430 "probe returned no data, skipping"
431 );
432 #[cfg(feature = "prometheus")]
433 Self::record_metric(&probe_name, "empty");
434 }
435 Err(e) => {
436 warn!(
437 probe = %probe_name,
438 error = %e,
439 "probe error, skipping this poll"
440 );
441 #[cfg(feature = "prometheus")]
442 Self::record_metric(&probe_name, "error");
443 }
444 }
445 }
446 }
447 }
448 })
449 }
450}
451
452#[cfg(test)]
453mod tests {
454 use std::sync::Arc;
455 use std::sync::atomic::{AtomicU32, Ordering};
456 use std::time::Duration;
457
458 use serde_json::json;
459 use tokio::time::timeout;
460 use tokio_util::sync::CancellationToken;
461
462 use super::*;
463
464 struct FixedProbe {
465 data: Value,
466 }
467
468 impl FixedProbe {
469 fn new(data: Value) -> Self {
470 Self { data }
471 }
472 }
473
474 impl PollingProbe for FixedProbe {
475 fn name(&self) -> &str {
476 "fixed"
477 }
478
479 fn poll(&self) -> ProbeFuture<'_> {
480 let data = self.data.clone();
481 Box::pin(async move { Ok(Some(ProbeResult::new(data))) })
482 }
483 }
484
485 struct CountingProbe {
486 counter: Arc<AtomicU32>,
487 }
488
489 impl CountingProbe {
490 fn new(counter: Arc<AtomicU32>) -> Self {
491 Self { counter }
492 }
493 }
494
495 impl PollingProbe for CountingProbe {
496 fn name(&self) -> &str {
497 "counting"
498 }
499
500 fn poll(&self) -> ProbeFuture<'_> {
501 let n = self.counter.fetch_add(1, Ordering::SeqCst);
502 Box::pin(async move { Ok(Some(ProbeResult::new(json!({"poll": n})))) })
503 }
504 }
505
506 struct EmptyProbe;
507
508 impl PollingProbe for EmptyProbe {
509 fn name(&self) -> &str {
510 "empty"
511 }
512
513 fn poll(&self) -> ProbeFuture<'_> {
514 Box::pin(async { Ok(None) })
515 }
516 }
517
518 struct ErrorProbe;
519
520 impl PollingProbe for ErrorProbe {
521 fn name(&self) -> &str {
522 "error"
523 }
524
525 fn poll(&self) -> ProbeFuture<'_> {
526 Box::pin(async { Err(ProbeError::Failed("connection refused".to_string())) })
527 }
528 }
529
530 fn make_trigger(probe: Box<dyn PollingProbe>, dedup: bool) -> PollingTrigger {
531 PollingTrigger::new(PollingTriggerConfig {
532 interval: Duration::from_millis(50),
533 probe,
534 workflow_name: "test-workflow".to_string(),
535 dedup,
536 })
537 }
538
539 #[tokio::test]
540 async fn polling_triggers_on_first_poll() {
541 let trigger = make_trigger(Box::new(FixedProbe::new(json!({"key": "val"}))), true);
542 let (sink, mut rx) = TriggerSink::channel(16);
543 let token = CancellationToken::new();
544 let token_clone = token.clone();
545
546 let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
547
548 let event = timeout(Duration::from_secs(2), rx.recv())
549 .await
550 .expect("timed out")
551 .expect("channel closed");
552
553 assert_eq!(event.workflow_name, "test-workflow");
554 assert_eq!(event.payload["key"], "val");
555 assert!(matches!(
556 event.trigger_kind,
557 TriggerKind::Polling { ref probe } if probe == "fixed"
558 ));
559
560 token.cancel();
561 let _ = handle.await;
562 }
563
564 #[tokio::test]
565 async fn polling_dedup_skips_unchanged() {
566 let trigger = make_trigger(Box::new(FixedProbe::new(json!({"static": true}))), true);
567 let (sink, mut rx) = TriggerSink::channel(16);
568 let token = CancellationToken::new();
569 let token_clone = token.clone();
570
571 let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
572
573 // First poll triggers
574 let _event = timeout(Duration::from_secs(2), rx.recv())
575 .await
576 .expect("timed out")
577 .expect("channel closed");
578
579 // Wait for a couple more poll cycles
580 tokio::time::sleep(Duration::from_millis(150)).await;
581
582 // No second event should arrive
583 assert!(rx.try_recv().is_err());
584
585 token.cancel();
586 let _ = handle.await;
587 }
588
589 #[tokio::test]
590 async fn polling_no_dedup_fires_every_time() {
591 let trigger = make_trigger(Box::new(FixedProbe::new(json!({"static": true}))), false);
592 let (sink, mut rx) = TriggerSink::channel(16);
593 let token = CancellationToken::new();
594 let token_clone = token.clone();
595
596 let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
597
598 // Should get at least 2 events even with identical data
599 let _e1 = timeout(Duration::from_secs(2), rx.recv())
600 .await
601 .expect("timed out")
602 .expect("channel closed");
603
604 let _e2 = timeout(Duration::from_secs(2), rx.recv())
605 .await
606 .expect("timed out")
607 .expect("channel closed");
608
609 token.cancel();
610 let _ = handle.await;
611 }
612
613 #[tokio::test]
614 async fn polling_error_does_not_trigger() {
615 let trigger = make_trigger(Box::new(ErrorProbe), false);
616 let (sink, mut rx) = TriggerSink::channel(16);
617 let token = CancellationToken::new();
618 let token_clone = token.clone();
619
620 let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
621
622 // Wait for a few poll cycles
623 tokio::time::sleep(Duration::from_millis(200)).await;
624
625 // No event should have been emitted
626 assert!(rx.try_recv().is_err());
627
628 token.cancel();
629 let _ = handle.await;
630 }
631
632 #[tokio::test]
633 async fn polling_graceful_shutdown() {
634 let trigger = make_trigger(Box::new(FixedProbe::new(json!(null))), false);
635 let (sink, _rx) = TriggerSink::channel(16);
636 let token = CancellationToken::new();
637 let token_clone = token.clone();
638
639 let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
640
641 tokio::time::sleep(Duration::from_millis(50)).await;
642 assert!(!handle.is_finished());
643
644 token.cancel();
645 let result = timeout(Duration::from_secs(2), handle)
646 .await
647 .expect("timed out")
648 .expect("task panicked");
649 assert!(result.is_ok());
650 }
651
652 #[tokio::test]
653 async fn polling_dedup_fires_on_change() {
654 let counter = Arc::new(AtomicU32::new(0));
655 let trigger = make_trigger(Box::new(CountingProbe::new(counter)), true);
656 let (sink, mut rx) = TriggerSink::channel(16);
657 let token = CancellationToken::new();
658 let token_clone = token.clone();
659
660 let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
661
662 // Every poll returns different data, so dedup should NOT block
663 let _e1 = timeout(Duration::from_secs(2), rx.recv())
664 .await
665 .expect("timed out")
666 .expect("channel closed");
667
668 let _e2 = timeout(Duration::from_secs(2), rx.recv())
669 .await
670 .expect("timed out")
671 .expect("channel closed");
672
673 token.cancel();
674 let _ = handle.await;
675 }
676
677 #[tokio::test]
678 async fn polling_empty_probe_does_not_trigger() {
679 let trigger = make_trigger(Box::new(EmptyProbe), false);
680 let (sink, mut rx) = TriggerSink::channel(16);
681 let token = CancellationToken::new();
682 let token_clone = token.clone();
683
684 let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
685
686 tokio::time::sleep(Duration::from_millis(200)).await;
687
688 assert!(rx.try_recv().is_err());
689
690 token.cancel();
691 let _ = handle.await;
692 }
693
694 #[test]
695 fn probe_result_hash_deterministic() {
696 let r1 = ProbeResult::new(json!({"a": 1, "b": 2}));
697 let r2 = ProbeResult::new(json!({"a": 1, "b": 2}));
698 assert_eq!(r1.content_hash(), r2.content_hash());
699 }
700
701 #[test]
702 fn probe_result_hash_differs_for_different_data() {
703 let r1 = ProbeResult::new(json!({"a": 1}));
704 let r2 = ProbeResult::new(json!({"a": 2}));
705 assert_ne!(r1.content_hash(), r2.content_hash());
706 }
707
708 #[test]
709 fn probe_result_with_custom_hash() {
710 let r = ProbeResult::with_hash(json!(null), "custom-hash".to_string());
711 assert_eq!(r.content_hash(), "custom-hash");
712 }
713
714 #[test]
715 fn probe_error_display() {
716 let err = ProbeError::Failed("timeout".to_string());
717 assert_eq!(err.to_string(), "probe failed: timeout");
718 }
719
720 #[cfg(feature = "prometheus")]
721 #[tokio::test]
722 async fn polling_metrics_incremented() {
723 use metrics_exporter_prometheus::PrometheusBuilder;
724
725 let recorder = PrometheusBuilder::new().build_recorder();
726 let prom_handle = recorder.handle();
727 metrics::set_global_recorder(recorder).ok();
728
729 let trigger = make_trigger(Box::new(FixedProbe::new(json!({"m": 1}))), false);
730 let (sink, mut rx) = TriggerSink::channel(16);
731 let token = CancellationToken::new();
732 let token_clone = token.clone();
733
734 let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
735
736 let _ = timeout(Duration::from_secs(2), rx.recv()).await;
737
738 token.cancel();
739 let _ = handle.await;
740
741 let output = prom_handle.render();
742 assert!(
743 output.contains("ironflow_polling_trigger_total"),
744 "expected polling metric in: {output}"
745 );
746 }
747}