aimdb_core/transform/
mod.rs

1//! Reactive transform primitives for derived records.
2//!
3//! # Transform Archetypes
4//!
5//! - **Map** (1:1, stateless): Transform each input value to zero-or-one output value
6//! - **Accumulate** (N:1, stateful): Aggregate a stream of values with persistent state
7//! - **Join** (M×N:1, stateful, multi-input): Combine values from multiple input records
8//!
9//! All three are handled by a unified API surface:
10//! - Single-input: `.transform()` with `TransformBuilder`
11//! - Multi-input: `.transform_join()` with `JoinBuilder`
12
13use core::fmt::Debug;
14
15use alloc::{boxed::Box, string::String, sync::Arc, vec::Vec};
16
17use crate::typed_record::BoxFuture;
18
19pub mod join;
20pub mod single;
21
22// Public re-exports
23pub use single::{StatefulTransformBuilder, TransformBuilder, TransformPipeline};
24
25#[cfg(feature = "observability")]
26use crate::{profiling::Clock, StageMetrics};
27pub use join::{JoinBuilder, JoinEventRx, JoinPipeline, JoinTrigger};
28
29// ============================================================================
30// TransformDescriptor — stored per output record in TypedRecord
31// ============================================================================
32
33/// Futures contributed by a transform at build-time collection.
34///
35/// `task_future` is the transform's main event loop; `fanin_futures` is any
36/// per-input forwarder needed for multi-input joins (empty for single-input).
37pub(crate) struct CollectedTransform {
38    pub task_future: BoxFuture<'static, ()>,
39    pub fanin_futures: Vec<BoxFuture<'static, ()>>,
40}
41
42#[cfg(feature = "observability")]
43pub(crate) type TransformProfiling = Option<(Arc<StageMetrics>, Clock)>;
44
45#[cfg(not(feature = "observability"))]
46pub(crate) type TransformProfiling = ();
47
48pub(crate) struct TransformDescriptor<T>
49where
50    T: Send + 'static + Debug + Clone,
51{
52    pub input_keys: Vec<String>,
53
54    /// Build the transform's futures.
55    ///
56    /// Receives:
57    /// - `Producer<T>` — the pre-resolved write handle for the output record.
58    /// - `Arc<AimDb>` — used by input forwarders to resolve their input
59    ///   consumers at startup time.
60    /// - `&str` — the output record's key, threaded through so the transform
61    ///   task can include it in tracing messages even though `Producer<T>` no
62    ///   longer carries a `.key()` accessor. Important when
63    ///   multiple records share type `T` under different keys.
64    #[allow(clippy::type_complexity)]
65    pub build_fn: Box<
66        dyn FnOnce(
67                crate::Producer<T>,
68                Arc<crate::AimDb>,
69                &str,
70                TransformProfiling,
71            ) -> CollectedTransform
72            + Send,
73    >,
74}
aimdb_core/transform/mod.rs

aimdb_core/transform/
mod.rs
