ops_rs/
macros.rs

1/// Store a variable in dry context using its name as the key
2/// dry_put!(dry_context, variable_name)
3#[macro_export]
4macro_rules! dry_put {
5    ($ctx:expr, $var:ident) => {
6        $ctx.insert(stringify!($var), $var)
7    };
8    ($ctx:expr, $var:ident, $value:expr) => {
9        $ctx.insert(stringify!($var), $value)
10    };
11}
12
13/// Store a value in dry context using a custom key name
14/// dry_put_key!(dry_context, "custom_key", value)
15#[macro_export]
16macro_rules! dry_put_key {
17    ($ctx:expr, $key:expr, $value:expr) => {
18        $ctx.insert($key, $value)
19    };
20}
21
22/// Retrieve a value from dry context using variable name as key
23/// let var: Type = dry_get!(dry_context, var_name);
24#[macro_export]
25macro_rules! dry_get {
26    ($ctx:expr, $var:ident) => {
27        $ctx.get::<_>(stringify!($var))
28    };
29}
30
31/// Retrieve a value from dry context using a custom key name
32/// let var: Option<Type> = dry_get_key!(dry_context, "custom_key");
33#[macro_export]
34macro_rules! dry_get_key {
35    ($ctx:expr, $key:expr) => {
36        $ctx.get::<_>($key)
37    };
38}
39
40/// Retrieve a required value from dry context, return error if missing
41/// let var: Type = dry_require!(dry_context, var_name)?;
42#[macro_export]
43macro_rules! dry_require {
44    ($ctx:expr, $var:ident) => {
45        $ctx.get_required::<_>(stringify!($var))
46    };
47}
48
49/// Retrieve a required value from dry context using a custom key name
50/// let var: Type = dry_require_key!(dry_context, "custom_key")?;
51#[macro_export]
52macro_rules! dry_require_key {
53    ($ctx:expr, $key:expr) => {
54        $ctx.get_required::<_>($key)
55    };
56}
57
58/// Store result in dry context under both the op name and "result" key
59/// dry_result!(dry_context, "OpName", result_value);
60#[macro_export]
61macro_rules! dry_result {
62    ($ctx:expr, $trigger_name:expr, $result:expr) => {{
63        $ctx.insert($trigger_name, $result.clone());
64        $ctx.insert("result", $result);
65    }};
66}
67
68/// Store a reference in wet context without serialization
69/// wet_put_ref!(wet_context, var_name, value);
70#[macro_export]
71macro_rules! wet_put_ref {
72    ($ctx:expr, $var:ident, $value:expr) => {
73        $ctx.insert_ref(stringify!($var), $value)
74    };
75    ($ctx:expr, $var:ident) => {
76        $ctx.insert_ref(stringify!($var), $var)
77    };
78}
79
80/// Store a reference in wet context using a custom key name
81/// wet_put_ref_key!(wet_context, "custom_key", value);
82#[macro_export]
83macro_rules! wet_put_ref_key {
84    ($ctx:expr, $key:expr, $value:expr) => {
85        $ctx.insert_ref($key, $value)
86    };
87}
88
89/// Store an Arc in wet context without additional wrapping
90/// wet_put_arc!(wet_context, var_name, arc_value);
91#[macro_export]
92macro_rules! wet_put_arc {
93    ($ctx:expr, $var:ident, $value:expr) => {
94        $ctx.insert_arc(stringify!($var), $value)
95    };
96    ($ctx:expr, $var:ident) => {
97        $ctx.insert_arc(stringify!($var), $var)
98    };
99}
100
101/// Store an Arc in wet context using a custom key name
102/// wet_put_arc_key!(wet_context, "custom_key", arc_value);
103#[macro_export]
104macro_rules! wet_put_arc_key {
105    ($ctx:expr, $key:expr, $value:expr) => {
106        $ctx.insert_arc($key, $value)
107    };
108}
109
110/// Get a reference from wet context
111/// let var: Option<Arc<Type>> = wet_get_ref!(wet_context, var_name);
112#[macro_export]
113macro_rules! wet_get_ref {
114    ($ctx:expr, $var:ident) => {
115        $ctx.get_ref::<_>(stringify!($var))
116    };
117}
118
119/// Get a reference from wet context using a custom key name
120/// let var: Option<Arc<Type>> = wet_get_ref_key!(wet_context, "custom_key");
121#[macro_export]
122macro_rules! wet_get_ref_key {
123    ($ctx:expr, $key:expr) => {
124        $ctx.get_ref::<_>($key)
125    };
126}
127
128/// Require a reference from wet context with error handling
129/// let var: Arc<Type> = wet_require_ref!(wet_context, var_name)?;
130#[macro_export]
131macro_rules! wet_require_ref {
132    ($ctx:expr, $var:ident) => {
133        $ctx.get_required::<_>(stringify!($var))
134    };
135}
136
137/// Require a reference from wet context using a custom key name
138/// let var: Arc<Type> = wet_require_ref_key!(wet_context, "custom_key")?;
139#[macro_export]
140macro_rules! wet_require_ref_key {
141    ($ctx:expr, $key:expr) => {
142        $ctx.get_required::<_>($key)
143    };
144}
145
146/// Aggregation strategies for composable ops
147#[derive(Debug, Clone)]
148pub enum AggregationStrategy {
149    /// Return all results as Vec<T>
150    All,
151    /// Return only the last result
152    Last,
153    /// Return only the first result
154    First,
155    /// Return the merged result (for types that support merging)
156    Merge,
157    /// Return unit () - ignore all results
158    Unit,
159}
160
161/// Create a named batch op type with flexible return types and aggregation
162/// Usage:
163/// ```rust,ignore
164/// // Returns Vec<T> (default behavior)
165/// batch! {
166///     ProcessingPipeline<String> = [
167///         ValidationOp::new(),
168///         TransformOp::new()
169///     ]
170/// }
171///
172/// // Returns T using last result
173/// batch! {
174///     ProcessingPipeline<String> -> last = [
175///         ValidationOp::new(),
176///         TransformOp::new()
177///     ]
178/// }
179///
180/// // Returns () ignoring all results
181/// batch! {
182///     ProcessingPipeline<()> -> unit = [
183///         SideEffectOp::new(),
184///         LoggingOp::new()
185///     ]
186/// }
187/// ```
188#[macro_export]
189macro_rules! batch {
190    // Default behavior: returns Vec<T>
191    ($name:ident<$T:ty> = [$($op:expr),+ $(,)?]) => {
192        batch!($name<$T> -> all = [$($op),+]);
193    };
194
195    // Flexible aggregation strategy
196    ($name:ident<$T:ty> -> $strategy:ident = [$($op:expr),+ $(,)?]) => {
197        #[derive(Debug, Clone)]
198        pub struct $name {
199            batch: $crate::BatchOp<$T>,
200            strategy: $crate::macros::AggregationStrategy,
201        }
202
203        impl $name {
204            pub fn new() -> Self {
205                let ops: Vec<std::sync::Arc<dyn $crate::Op<$T>>> = vec![
206                    $(std::sync::Arc::new($op)),+
207                ];
208                Self {
209                    batch: $crate::BatchOp::new(ops),
210                    strategy: batch!(@strategy $strategy),
211                }
212            }
213
214            pub fn with_continue_on_error(mut self, continue_on_error: bool) -> Self {
215                self.batch = self.batch.with_continue_on_error(continue_on_error);
216                self
217            }
218        }
219
220        impl Default for $name {
221            fn default() -> Self {
222                Self::new()
223            }
224        }
225
226        batch!(@impl $name<$T> -> $strategy);
227    };
228
229    // Strategy selection helper
230    (@strategy all) => { $crate::macros::AggregationStrategy::All };
231    (@strategy last) => { $crate::macros::AggregationStrategy::Last };
232    (@strategy first) => { $crate::macros::AggregationStrategy::First };
233    (@strategy merge) => { $crate::macros::AggregationStrategy::Merge };
234    (@strategy unit) => { $crate::macros::AggregationStrategy::Unit };
235
236    // Implementation for Vec<T> return (all strategy)
237    (@impl $name:ident<$T:ty> -> all) => {
238        #[async_trait::async_trait]
239        impl $crate::Op<Vec<$T>> for $name {
240            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<Vec<$T>> {
241                self.batch.perform(dry, wet).await
242            }
243
244            fn metadata(&self) -> $crate::OpMetadata {
245                self.batch.metadata()
246            }
247        }
248    };
249
250    // Implementation for T return (last strategy)
251    (@impl $name:ident<$T:ty> -> last) => {
252        #[async_trait::async_trait]
253        impl $crate::Op<$T> for $name {
254            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<$T> {
255                let results = self.batch.perform(dry, wet).await?;
256                results.into_iter().last()
257                    .ok_or_else(|| $crate::OpError::ExecutionFailed("No results from batch operation".to_string()))
258            }
259
260            fn metadata(&self) -> $crate::OpMetadata {
261                self.batch.metadata()
262            }
263        }
264    };
265
266    // Implementation for T return (first strategy)
267    (@impl $name:ident<$T:ty> -> first) => {
268        #[async_trait::async_trait]
269        impl $crate::Op<$T> for $name {
270            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<$T> {
271                let results = self.batch.perform(dry, wet).await?;
272                results.into_iter().next()
273                    .ok_or_else(|| $crate::OpError::ExecutionFailed("No results from batch operation".to_string()))
274            }
275
276            fn metadata(&self) -> $crate::OpMetadata {
277                self.batch.metadata()
278            }
279        }
280    };
281
282    // Implementation for () return (unit strategy)
283    (@impl $name:ident<$T:ty> -> unit) => {
284        #[async_trait::async_trait]
285        impl $crate::Op<()> for $name {
286            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<()> {
287                let _results = self.batch.perform(dry, wet).await?;
288                Ok(())
289            }
290
291            fn metadata(&self) -> $crate::OpMetadata {
292                self.batch.metadata()
293            }
294        }
295    };
296
297    // Implementation for merge strategy (requires Default + Clone)
298    (@impl $name:ident<$T:ty> -> merge) => {
299        #[async_trait::async_trait]
300        impl $crate::Op<$T> for $name
301        where
302            $T: Default + Clone
303        {
304            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<$T> {
305                let results = self.batch.perform(dry, wet).await?;
306                // Simple merge strategy - for more complex merging, implement custom logic
307                Ok(results.into_iter().last().unwrap_or_default())
308            }
309
310            fn metadata(&self) -> $crate::OpMetadata {
311                self.batch.metadata()
312            }
313        }
314    };
315}
316
317/// Create a named for-loop op type with flexible return types
318/// The limit is loaded from a variable in the dry context
319/// Usage:
320/// ```rust,ignore
321/// // Returns Vec<T> (default)
322/// repeat! {
323///     ProcessLoop<String> = {
324///         counter: "iteration",
325///         limit: "max_iterations",
326///         ops: [StepOp::new(), LogOp::new()]
327///     }
328/// }
329///
330/// // Returns T using last result
331/// repeat! {
332///     ProcessLoop<String> -> last = {
333///         counter: "iteration",
334///         limit: "max_iterations",
335///         ops: [StepOp::new(), LogOp::new()]
336///     }
337/// }
338/// ```
339#[macro_export]
340macro_rules! repeat {
341    // Default behavior: returns Vec<T>
342    ($name:ident<$T:ty> = {
343        counter: $counter:expr,
344        limit: $limit_var:expr,
345        ops: [$($op:expr),+ $(,)?]
346    }) => {
347        repeat!($name<$T> -> all = {
348            counter: $counter,
349            limit: $limit_var,
350            ops: [$($op),+]
351        });
352    };
353
354    // With continue_on_error option
355    ($name:ident<$T:ty> = {
356        counter: $counter:expr,
357        limit: $limit_var:expr,
358        continue_on_error: $continue_on_error:expr,
359        ops: [$($op:expr),+ $(,)?]
360    }) => {
361        repeat!($name<$T> -> all = {
362            counter: $counter,
363            limit: $limit_var,
364            continue_on_error: $continue_on_error,
365            ops: [$($op),+]
366        });
367    };
368
369    // Flexible aggregation strategy
370    ($name:ident<$T:ty> -> $strategy:ident = {
371        counter: $counter:expr,
372        limit: $limit_var:expr,
373        ops: [$($op:expr),+ $(,)?]
374    }) => {
375        repeat!($name<$T> -> $strategy = {
376            counter: $counter,
377            limit: $limit_var,
378            continue_on_error: false,
379            ops: [$($op),+]
380        });
381    };
382
383    // Flexible aggregation strategy with continue_on_error
384    ($name:ident<$T:ty> -> $strategy:ident = {
385        counter: $counter:expr,
386        limit: $limit_var:expr,
387        continue_on_error: $continue_on_error:expr,
388        ops: [$($op:expr),+ $(,)?]
389    }) => {
390        #[derive(Debug, Clone)]
391        pub struct $name {
392            counter_var: String,
393            limit_var: String,
394            continue_on_error: bool,
395            strategy: $crate::macros::AggregationStrategy,
396        }
397
398        impl $name {
399            pub fn new() -> Self {
400                Self {
401                    counter_var: $counter.to_string(),
402                    limit_var: $limit_var.to_string(),
403                    continue_on_error: $continue_on_error,
404                    strategy: repeat!(@strategy $strategy),
405                }
406            }
407
408            fn create_ops() -> Vec<std::sync::Arc<dyn $crate::Op<$T>>> {
409                vec![
410                    $(std::sync::Arc::new($op)),+
411                ]
412            }
413        }
414
415        impl Default for $name {
416            fn default() -> Self {
417                Self::new()
418            }
419        }
420
421        repeat!(@impl $name<$T> -> $strategy);
422
423        impl $name {
424            fn build_metadata(&self) -> $crate::OpMetadata {
425                let limit_var = self.limit_var.clone();
426                let mut properties = serde_json::Map::new();
427                properties.insert(
428                    limit_var.clone(),
429                    serde_json::json!({ "type": "integer", "minimum": 0 })
430                );
431
432                $crate::OpMetadata::builder(stringify!($name))
433                    .description(format!("For-loop with dynamic limit from '{}'", limit_var))
434                    .input_schema(serde_json::json!({
435                        "type": "object",
436                        "properties": properties,
437                        "required": [limit_var]
438                    }))
439                    .build()
440            }
441        }
442    };
443
444    // Strategy selection helper
445    (@strategy all) => { $crate::macros::AggregationStrategy::All };
446    (@strategy last) => { $crate::macros::AggregationStrategy::Last };
447    (@strategy first) => { $crate::macros::AggregationStrategy::First };
448    (@strategy merge) => { $crate::macros::AggregationStrategy::Merge };
449    (@strategy unit) => { $crate::macros::AggregationStrategy::Unit };
450
451    // Implementation for Vec<T> return (all strategy)
452    (@impl $name:ident<$T:ty> -> all) => {
453        #[async_trait::async_trait]
454        impl $crate::Op<Vec<$T>> for $name {
455            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<Vec<$T>> {
456                let limit = dry.get_required::<usize>(&self.limit_var)?;
457                let ops = Self::create_ops();
458                let loop_op = $crate::loop_op::LoopOp::new(
459                    self.counter_var.clone(),
460                    limit,
461                    ops
462                ).with_continue_on_error(self.continue_on_error);
463                loop_op.perform(dry, wet).await
464            }
465
466            fn metadata(&self) -> $crate::OpMetadata {
467                self.build_metadata()
468            }
469        }
470    };
471
472    // Implementation for T return (last strategy)
473    (@impl $name:ident<$T:ty> -> last) => {
474        #[async_trait::async_trait]
475        impl $crate::Op<$T> for $name {
476            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<$T> {
477                let limit = dry.get_required::<usize>(&self.limit_var)?;
478                let ops = Self::create_ops();
479                let loop_op = $crate::loop_op::LoopOp::new(
480                    self.counter_var.clone(),
481                    limit,
482                    ops
483                ).with_continue_on_error(self.continue_on_error);
484                let results = loop_op.perform(dry, wet).await?;
485                results.into_iter().last()
486                    .ok_or_else(|| $crate::OpError::ExecutionFailed("No results from repeat operation".to_string()))
487            }
488
489            fn metadata(&self) -> $crate::OpMetadata {
490                self.build_metadata()
491            }
492        }
493    };
494
495    // Implementation for T return (first strategy)
496    (@impl $name:ident<$T:ty> -> first) => {
497        #[async_trait::async_trait]
498        impl $crate::Op<$T> for $name {
499            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<$T> {
500                let limit = dry.get_required::<usize>(&self.limit_var)?;
501                let ops = Self::create_ops();
502                let loop_op = $crate::loop_op::LoopOp::new(
503                    self.counter_var.clone(),
504                    limit,
505                    ops
506                ).with_continue_on_error(self.continue_on_error);
507                let results = loop_op.perform(dry, wet).await?;
508                results.into_iter().next()
509                    .ok_or_else(|| $crate::OpError::ExecutionFailed("No results from repeat operation".to_string()))
510            }
511
512            fn metadata(&self) -> $crate::OpMetadata {
513                self.build_metadata()
514            }
515        }
516    };
517
518    // Implementation for () return (unit strategy)
519    (@impl $name:ident<$T:ty> -> unit) => {
520        #[async_trait::async_trait]
521        impl $crate::Op<()> for $name {
522            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<()> {
523                let limit = dry.get_required::<usize>(&self.limit_var)?;
524                let ops = Self::create_ops();
525                let loop_op = $crate::loop_op::LoopOp::new(
526                    self.counter_var.clone(),
527                    limit,
528                    ops
529                ).with_continue_on_error(self.continue_on_error);
530                let _results = loop_op.perform(dry, wet).await?;
531                Ok(())
532            }
533
534            fn metadata(&self) -> $crate::OpMetadata {
535                self.build_metadata()
536            }
537        }
538    };
539
540    // Implementation for merge strategy
541    (@impl $name:ident<$T:ty> -> merge) => {
542        #[async_trait::async_trait]
543        impl $crate::Op<$T> for $name
544        where
545            $T: Default + Clone
546        {
547            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<$T> {
548                let limit = dry.get_required::<usize>(&self.limit_var)?;
549                let ops = Self::create_ops();
550                let loop_op = $crate::loop_op::LoopOp::new(
551                    self.counter_var.clone(),
552                    limit,
553                    ops
554                ).with_continue_on_error(self.continue_on_error);
555                let results = loop_op.perform(dry, wet).await?;
556                Ok(results.into_iter().last().unwrap_or_default())
557            }
558
559            fn metadata(&self) -> $crate::OpMetadata {
560                self.build_metadata()
561            }
562        }
563    };
564}
565
566/// Create a named while-loop op type with condition-based iteration
567/// The condition is checked from a boolean variable in the dry context
568/// Internal control flags use unique UUIDs to prevent conflicts in nested loops
569/// Usage:
570/// ```rust,ignore
571/// // Returns Vec<T> (default)
572/// repeat_until! {
573///     ProcessUntilDone<String> = {
574///         counter: "iteration",
575///         condition: "should_continue",  // Boolean variable in dry context
576///         max_iterations: 100,           // Safety limit
577///         ops: [ProcessOp::new(), CheckOp::new()]
578///     }
579/// }
580///
581/// // Returns T using last result
582/// repeat_until! {
583///     ProcessUntilDone<String> -> last = {
584///         counter: "iteration",
585///         condition: "should_continue",
586///         max_iterations: 100,
587///         ops: [ProcessOp::new(), CheckOp::new()]
588///     }
589/// }
590/// ```
591#[macro_export]
592macro_rules! repeat_until {
593    // Default behavior: returns Vec<T>
594    ($name:ident<$T:ty> = {
595        counter: $counter:expr,
596        condition: $condition_var:expr,
597        max_iterations: $max:expr,
598        ops: [$($op:expr),+ $(,)?]
599    }) => {
600        repeat_until!($name<$T> -> all = {
601            counter: $counter,
602            condition: $condition_var,
603            max_iterations: $max,
604            ops: [$($op),+]
605        });
606    };
607
608    // Flexible aggregation strategy
609    ($name:ident<$T:ty> -> $strategy:ident = {
610        counter: $counter:expr,
611        condition: $condition_var:expr,
612        max_iterations: $max:expr,
613        ops: [$($op:expr),+ $(,)?]
614    }) => {
615        #[derive(Debug, Clone)]
616        pub struct $name {
617            counter_var: String,
618            condition_var: String,
619            max_iterations: usize,
620            strategy: $crate::macros::AggregationStrategy,
621            // Unique control variables to prevent conflicts in nested loops
622            continue_var: String,
623            break_var: String,
624            loop_id: String,
625        }
626
627        impl $name {
628            pub fn new() -> Self {
629                // Generate a unique loop ID for scoped control flow variables
630				// Keep the preceding double underscores to minimize risk of collisions
631				let loop_id = ::uuid::Uuid::new_v4().to_string();
632                Self {
633                    counter_var: $counter.to_string(),
634                    condition_var: $condition_var.to_string(),
635                    max_iterations: $max,
636                    strategy: repeat_until!(@strategy $strategy),
637                    continue_var: format!("__continue_loop_{}", loop_id),
638                    break_var: format!("__break_loop_{}", loop_id),
639                    loop_id,
640                }
641            }
642
643            fn create_ops() -> Vec<std::sync::Arc<dyn $crate::Op<$T>>> {
644                vec![
645                    $(std::sync::Arc::new($op)),+
646                ]
647            }
648
649            async fn perform_repeat_until(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<Vec<$T>> {
650                let mut results = Vec::new();
651                let mut iteration = 0;
652
653                // Set loop context for nested control flow
654                dry.insert("__current_loop_id", &self.loop_id);
655
656                while iteration < self.max_iterations {
657                    // Check condition from context
658                    let should_continue = dry.get::<bool>(&self.condition_var)
659                        .unwrap_or(false);
660
661                    if !should_continue {
662                        break;
663                    }
664
665                    // Set counter in context
666                    dry.insert(&self.counter_var, iteration);
667
668                    // Clear scoped control flags for this iteration
669                    dry.insert(&self.continue_var, false);
670                    dry.insert(&self.break_var, false);
671
672                    // Execute ops
673                    let ops = Self::create_ops();
674                    for op in ops {
675                        let result = op.perform(dry, wet).await?;
676                        results.push(result);
677
678                        // Check for early termination signals
679                        if dry.is_aborted() {
680                            return Err($crate::OpError::Aborted(
681                                dry.abort_reason().cloned().unwrap_or_else(|| "While loop aborted".to_string())
682                            ));
683                        }
684
685                        // Check scoped continue flag
686                        if dry.get::<bool>(&self.continue_var).unwrap_or(false) {
687                            dry.insert(&self.continue_var, false); // Clear flag
688                            break; // Continue to next iteration
689                        }
690
691                        // Check scoped break flag
692                        if dry.get::<bool>(&self.break_var).unwrap_or(false) {
693                            dry.insert(&self.break_var, false); // Clear flag
694                            return Ok(results); // Break out of entire loop
695                        }
696                    }
697
698                    iteration += 1;
699                }
700
701                Ok(results)
702            }
703        }
704
705        impl Default for $name {
706            fn default() -> Self {
707                Self::new()
708            }
709        }
710
711        repeat_until!(@impl $name<$T> -> $strategy);
712
713        impl $name {
714            fn build_metadata(&self) -> $crate::OpMetadata {
715                let condition_var = self.condition_var.clone();
716                let mut properties = serde_json::Map::new();
717                properties.insert(
718                    condition_var.clone(),
719                    serde_json::json!({ "type": "boolean" })
720                );
721
722                $crate::OpMetadata::builder(stringify!($name))
723                    .description(format!("While-loop with condition from '{}' (max {} iterations)", condition_var, self.max_iterations))
724                    .input_schema(serde_json::json!({
725                        "type": "object",
726                        "properties": properties,
727                        "required": [condition_var]
728                    }))
729                    .build()
730            }
731        }
732    };
733
734    // Strategy selection helper
735    (@strategy all) => { $crate::macros::AggregationStrategy::All };
736    (@strategy last) => { $crate::macros::AggregationStrategy::Last };
737    (@strategy first) => { $crate::macros::AggregationStrategy::First };
738    (@strategy merge) => { $crate::macros::AggregationStrategy::Merge };
739    (@strategy unit) => { $crate::macros::AggregationStrategy::Unit };
740
741    // Implementation for Vec<T> return (all strategy)
742    (@impl $name:ident<$T:ty> -> all) => {
743        #[async_trait::async_trait]
744        impl $crate::Op<Vec<$T>> for $name {
745            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<Vec<$T>> {
746                self.perform_repeat_until(dry, wet).await
747            }
748
749            fn metadata(&self) -> $crate::OpMetadata {
750                self.build_metadata()
751            }
752        }
753    };
754
755    // Implementation for T return (last strategy)
756    (@impl $name:ident<$T:ty> -> last) => {
757        #[async_trait::async_trait]
758        impl $crate::Op<$T> for $name {
759            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<$T> {
760                let results = self.perform_repeat_until(dry, wet).await?;
761                results.into_iter().last()
762                    .ok_or_else(|| $crate::OpError::ExecutionFailed("No results from while loop operation".to_string()))
763            }
764
765            fn metadata(&self) -> $crate::OpMetadata {
766                self.build_metadata()
767            }
768        }
769    };
770
771    // Implementation for T return (first strategy)
772    (@impl $name:ident<$T:ty> -> first) => {
773        #[async_trait::async_trait]
774        impl $crate::Op<$T> for $name {
775            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<$T> {
776                let results = self.perform_repeat_until(dry, wet).await?;
777                results.into_iter().next()
778                    .ok_or_else(|| $crate::OpError::ExecutionFailed("No results from while loop operation".to_string()))
779            }
780
781            fn metadata(&self) -> $crate::OpMetadata {
782                self.build_metadata()
783            }
784        }
785    };
786
787    // Implementation for () return (unit strategy)
788    (@impl $name:ident<$T:ty> -> unit) => {
789        #[async_trait::async_trait]
790        impl $crate::Op<()> for $name {
791            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<()> {
792                let _results = self.perform_repeat_until(dry, wet).await?;
793                Ok(())
794            }
795
796            fn metadata(&self) -> $crate::OpMetadata {
797                self.build_metadata()
798            }
799        }
800    };
801
802    // Implementation for merge strategy
803    (@impl $name:ident<$T:ty> -> merge) => {
804        #[async_trait::async_trait]
805        impl $crate::Op<$T> for $name
806        where
807            $T: Default + Clone
808        {
809            async fn perform(&self, dry: &mut $crate::DryContext, wet: &mut $crate::WetContext) -> $crate::OpResult<$T> {
810                let results = self.perform_repeat_until(dry, wet).await?;
811                Ok(results.into_iter().last().unwrap_or_default())
812            }
813
814            fn metadata(&self) -> $crate::OpMetadata {
815                self.build_metadata()
816            }
817        }
818    };
819}
820
821/// Create a wrapper that converts any Op<T> to Op<U> using a conversion strategy
822/// For advanced composability between different op types
823/// Usage is typically not needed with the new flexible batch!/repeat!/repeat_until! macros
824#[macro_export]
825macro_rules! op_wrapper {
826    // Convert any Op<T> to Op<()> by ignoring the result
827    (unit $name:ident for $from:ty) => {
828        pub struct $name<T>
829        where
830            T: $crate::Op<$from> + Send + Sync,
831        {
832            inner: T,
833        }
834
835        impl<T> $name<T>
836        where
837            T: $crate::Op<$from> + Send + Sync,
838        {
839            pub fn new(op: T) -> Self {
840                Self { inner: op }
841            }
842        }
843
844        #[async_trait::async_trait]
845        impl<T> $crate::Op<()> for $name<T>
846        where
847            T: $crate::Op<$from> + Send + Sync,
848        {
849            async fn perform(
850                &self,
851                dry: &mut $crate::DryContext,
852                wet: &mut $crate::WetContext,
853            ) -> $crate::OpResult<()> {
854                let _result = self.inner.perform(dry, wet).await?;
855                Ok(())
856            }
857
858            fn metadata(&self) -> $crate::OpMetadata {
859                self.inner.metadata()
860            }
861        }
862    };
863
864    // Convert Op<Vec<T>> to Op<T> using last element
865    (last $name:ident for $elem:ty) => {
866        pub struct $name<T>
867        where
868            T: $crate::Op<Vec<$elem>> + Send + Sync,
869        {
870            inner: T,
871            _phantom: std::marker::PhantomData<$elem>,
872        }
873
874        impl<T> $name<T>
875        where
876            T: $crate::Op<Vec<$elem>> + Send + Sync,
877        {
878            pub fn new(op: T) -> Self {
879                Self {
880                    inner: op,
881                    _phantom: std::marker::PhantomData,
882                }
883            }
884        }
885
886        #[async_trait::async_trait]
887        impl<T> $crate::Op<$elem> for $name<T>
888        where
889            T: $crate::Op<Vec<$elem>> + Send + Sync,
890        {
891            async fn perform(
892                &self,
893                dry: &mut $crate::DryContext,
894                wet: &mut $crate::WetContext,
895            ) -> $crate::OpResult<$elem> {
896                let results = self.inner.perform(dry, wet).await?;
897                results.into_iter().last().ok_or_else(|| {
898                    $crate::OpError::ExecutionFailed(
899                        "No results from wrapped operation".to_string(),
900                    )
901                })
902            }
903
904            fn metadata(&self) -> $crate::OpMetadata {
905                self.inner.metadata()
906            }
907        }
908    };
909}
910
911// Control flow macros
912
913/// Abort macro that sets the abort flag in DryContext
914///
915/// This macro should be used when an operation determines that continuing
916/// execution would be futile and should not trigger retries.
917///
918/// # Usage
919/// ```
920/// use ops::{abort, DryContext, OpError};
921///
922/// fn example_op(dry: &mut DryContext) -> Result<(), OpError> {
923///     abort!(dry);
924/// }
925/// ```
926#[macro_export]
927macro_rules! abort {
928    ($dry:expr) => {{
929        $dry.set_abort(None);
930        return Err($crate::OpError::Aborted("Operation aborted".to_string()));
931    }};
932    ($dry:expr, $reason:expr) => {{
933        let reason_str = $reason.to_string();
934        $dry.set_abort(Some(reason_str.clone()));
935        return Err($crate::OpError::Aborted(reason_str));
936    }};
937}
938
939/// Continue loop macro that sets the continue flag in DryContext
940///
941/// This macro should be used within loop operations to skip the rest
942/// of the current iteration, similar to 'continue' in a for loop.
943///
944/// Automatically targets the current loop context.
945///
946/// # Usage
947/// ```
948/// use ops::{continue_loop, DryContext, OpError};
949///
950/// fn example_op(dry: &mut DryContext) -> Result<i32, OpError> {
951///     continue_loop!(dry);
952/// }
953/// ```
954#[macro_export]
955macro_rules! continue_loop {
956    ($dry:expr) => {{
957        // Get current loop ID and set scoped continue flag
958        if let Some(loop_id) = $dry.get::<String>("__current_loop_id") {
959            let continue_var = format!("__continue_loop_{}", loop_id);
960            $dry.insert(&continue_var, true);
961        }
962        return Ok(Default::default()); // Return default value for the op type
963    }};
964}
965
966/// Break loop macro that sets the break flag in DryContext
967///
968/// This macro should be used within loop operations to exit the entire
969/// loop immediately, similar to 'break' in a for loop.
970///
971/// Automatically targets the current loop context.
972///
973/// # Usage
974/// ```
975/// use ops::{break_loop, DryContext, OpError};
976///
977/// fn example_op(dry: &mut DryContext) -> Result<i32, OpError> {
978///     break_loop!(dry);
979/// }
980/// ```
981#[macro_export]
982macro_rules! break_loop {
983    ($dry:expr) => {{
984        // Get current loop ID and set scoped break flag
985        if let Some(loop_id) = $dry.get::<String>("__current_loop_id") {
986            let break_var = format!("__break_loop_{}", loop_id);
987            $dry.insert(&break_var, true);
988        }
989        return Ok(Default::default()); // Return default value for the op type
990    }};
991}
992
993/// Break a specific loop by targeting its loop ID
994///
995/// # Usage
996/// ```
997/// use ops::{break_loop_scoped, DryContext, OpError};
998///
999/// fn example_op(dry: &mut DryContext) -> Result<i32, OpError> {
1000///     break_loop_scoped!(dry, "my_loop_id");
1001/// }
1002/// ```
1003#[macro_export]
1004macro_rules! break_loop_scoped {
1005    ($dry:expr, $loop_id:expr) => {{
1006        let break_var = format!("__break_loop_{}", $loop_id);
1007        $dry.insert(&break_var, true);
1008        return Ok(Default::default()); // Return default value for the op type
1009    }};
1010}
1011
1012/// Utility macro to check if operation should be aborted
1013///
1014/// Returns early with Aborted error if abort flag is set
1015///
1016/// # Usage
1017/// ```
1018/// use ops::{check_abort, DryContext, OpError};
1019///
1020/// fn example_op(dry: &mut DryContext) -> Result<i32, OpError> {
1021///     check_abort!(dry);
1022///     Ok(42)
1023/// }
1024/// ```
1025#[macro_export]
1026macro_rules! check_abort {
1027    ($dry:expr) => {
1028        if $dry.is_aborted() {
1029            let reason = $dry
1030                .abort_reason()
1031                .cloned()
1032                .unwrap_or_else(|| "Operation aborted".to_string());
1033            return Err($crate::OpError::Aborted(reason));
1034        }
1035    };
1036}
1037
1038/// Macro to create a void wrapper type that discards the result
1039#[macro_export]
1040macro_rules! void_op {
1041    ($wrapper_name:ident, $op_type:ty) => {
1042        #[derive(Clone)]
1043        pub struct $wrapper_name {
1044            wrapped: $op_type,
1045        }
1046
1047        impl $wrapper_name {
1048            pub fn new() -> Self {
1049                Self {
1050                    wrapped: <$op_type>::new(),
1051                }
1052            }
1053        }
1054
1055        #[async_trait::async_trait]
1056        impl $crate::Op<()> for $wrapper_name {
1057            async fn perform(&self, dry: &mut DryContext, wet: &mut WetContext) -> OpResult<()> {
1058                self.wrapped.perform(dry, wet).await.map(|_| ())
1059            }
1060
1061            fn metadata(&self) -> OpMetadata {
1062                self.wrapped.metadata()
1063            }
1064
1065            async fn rollback(&self, dry: &mut DryContext, wet: &mut WetContext) -> OpResult<()> {
1066                self.wrapped.rollback(dry, wet).await
1067            }
1068        }
1069    };
1070}
1071
1072/// Macro to create a Trigger for any operation type
1073///
1074/// This macro generates all the boilerplate code needed to set a trigger for any Op type
1075/// and make it compatible with the TriggerRegistry system.
1076///
1077/// Usage:
1078/// ```rust,ignore
1079/// // Basic usage - creates MyTrigger struct
1080/// wire_trigger!(MyTrigger, MyOp, "MyOperation");
1081///
1082/// // Real examples
1083/// wire_trigger!(FolderScanTrigger, crate::ops::FolderScanOp, "FolderScanOp");
1084/// wire_trigger!(CustomAnalysisTrigger, MyCustomAnalysisOp, "CustomAnalysis");
1085///
1086/// // Then register in your registry
1087/// registry.register("my_task_type", || {
1088///     Box::new(MyTrigger::new())
1089/// });
1090/// ```
1091#[macro_export]
1092macro_rules! wire_trigger {
1093    ($wrapper_name:ident, $op_type:ty, $name:expr) => {
1094        paste::paste! {
1095            // Create a void wrapper for the op type with unique name
1096            $crate::void_op!([<$wrapper_name VoidOp>], $op_type);
1097
1098            pub struct $wrapper_name {
1099                name: String,
1100                action: [<$wrapper_name VoidOp>],
1101            }
1102
1103            impl $wrapper_name {
1104                pub fn new() -> Self {
1105                    Self {
1106                        name: $name.to_string(),
1107                        action: [<$wrapper_name VoidOp>]::new(),
1108                    }
1109                }
1110            }
1111        }
1112
1113        impl Default for $wrapper_name {
1114            fn default() -> Self {
1115                Self::new()
1116            }
1117        }
1118
1119        #[async_trait::async_trait]
1120        impl $crate::Trigger for $wrapper_name {
1121            fn name(&self) -> String {
1122                self.name.clone()
1123            }
1124
1125            fn actions(&self) -> Vec<Arc<dyn $crate::Op<()>>> {
1126                vec![Arc::new(self.action.clone())]
1127            }
1128
1129            fn predicate(&self) -> Arc<dyn $crate::Op<bool>> {
1130                Arc::new($crate::TrivialPredicate::default())
1131            }
1132        }
1133    };
1134
1135    ($wrapper_name:ident, $op_type:ty, $name:expr, $predicate:expr) => {
1136        paste::paste! {
1137            // Create a void wrapper for the op type with unique name
1138            $crate::void_op!([<$wrapper_name VoidOp>], $op_type);
1139
1140            pub struct $wrapper_name {
1141                name: String,
1142                action: [<$wrapper_name VoidOp>],
1143                predicate: Arc<dyn $crate::Op<bool>>,
1144            }
1145
1146            impl $wrapper_name {
1147                pub fn new() -> Self {
1148                    Self {
1149                        name: $name.to_string(),
1150                        action: [<$wrapper_name VoidOp>]::new(),
1151                        predicate: Arc::new($predicate),
1152                    }
1153                }
1154
1155                pub fn with_predicate(name: String, predicate: Arc<dyn $crate::Op<bool>>) -> Self {
1156                    Self {
1157                        name,
1158                        action: [<$wrapper_name VoidOp>]::new(),
1159                        predicate,
1160                    }
1161                }
1162            }
1163        }
1164
1165        impl Default for $wrapper_name {
1166            fn default() -> Self {
1167                Self::new()
1168            }
1169        }
1170
1171        #[async_trait::async_trait]
1172        impl $crate::Trigger for $wrapper_name {
1173            fn name(&self) -> String {
1174                self.name.clone()
1175            }
1176            fn predicate(&self) -> Arc<dyn $crate::Op<bool>> {
1177                Arc::clone(&self.predicate)
1178            }
1179            fn actions(&self) -> Vec<Arc<dyn $crate::Op<()>>> {
1180                vec![Arc::new(self.action.clone())]
1181            }
1182        }
1183    };
1184}
ops_rs/macros.rs

ops_rs/
macros.rs
