pulse_client/
streams.rs

1//! B-107 — Kafka-Streams-like declarative DSL that compiles to a Pulse pipeline.
2//!
3//! The DSL is **server-side execution, client-side declaration**: the operator
4//! chain is built in Rust, compiled to the JSON pipeline shape that the Pulse
5//! server's `StreamingOperatorValidator` accepts, and POSTed to
6//! `/api/pulse/pipelines`. Stream processing then runs on the Pulse engine
7//! (3.6 M evt/s native throughput), not in the client process.
8//!
9//! This is the opposite of Kafka Streams (which executes in the caller's JVM).
10//! The trade-off: you can't do microsecond client-side compute, but you get
11//! infinite-scale stateful streaming, durable replicated state queryable via
12//! B-106 IQ, and the same DSL works from any of the 5 Pulse SDKs.
13//!
14//! # Quick start
15//!
16//! ```no_run
17//! use pulse_client::{aggs, windows, PulseClient, StreamBuilder};
18//!
19//! # async fn run() -> Result<(), pulse_client::PulseError> {
20//! let client = PulseClient::builder()
21//!     .base_url("http://localhost:9090")
22//!     .token("ey...")
23//!     .build()?;
24//!
25//! let mut aggregations = std::collections::BTreeMap::new();
26//! aggregations.insert("avgTemp".to_string(), aggs::avg("temperature"));
27//!
28//! let builder = StreamBuilder::new("iot-temperature-aggregator")
29//!     .from_topic_with_engine("sensor-readings", "mqtt")
30//!     .key_by("deviceId")
31//!     .window_with_aggs(windows::tumbling("60s"), aggregations)
32//!     .filter("avgTemp > 75")
33//!     .to_topic_with_channel("sensor-minute-averages", "email");
34//!
35//! client.streams().deploy(&builder).await?;
36//! # Ok(())
37//! # }
38//! ```
39//!
40//! Supported operators (mirror the 11 validated by the server's
41//! `StreamingOperatorValidator`): `filter`, `map`, `flat_map`, `key_by`,
42//! `window`, `branch`, `enrich`, `enrich_async`, `cep`, `broadcast_join`,
43//! `cdc_join`.
44//!
45//! Conditions and field-expressions are passed as **strings** — closures /
46//! lambdas are NOT supported because they can't be serialised to JSON.
47
48use std::collections::BTreeMap;
49
50use reqwest::Method;
51use serde_json::{json, Map, Value};
52
53use crate::client::PulseClient;
54use crate::error::PulseError;
55
56// ---------------------------------------------------------------------------
57// Window specs
58// ---------------------------------------------------------------------------
59
60/// A window specification. Compiled to the string form the server expects.
61///
62/// Construct via the [`windows`] helpers — never instantiate directly unless
63/// you've validated the raw string against `WindowEngine.parseSpec`.
64#[derive(Debug, Clone, PartialEq, Eq, Hash)]
65pub struct WindowSpec {
66    spec: String,
67}
68
69impl WindowSpec {
70    /// Wraps a pre-validated raw spec string. Panics on empty input.
71    pub fn new(spec: impl Into<String>) -> Self {
72        let spec = spec.into();
73        if spec.trim().is_empty() {
74            panic!("WindowSpec requires a non-empty spec string");
75        }
76        Self { spec }
77    }
78
79    /// The raw spec string as it will appear on the wire.
80    pub fn spec(&self) -> &str {
81        &self.spec
82    }
83}
84
85impl std::fmt::Display for WindowSpec {
86    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
87        f.write_str(&self.spec)
88    }
89}
90
91/// Window-spec factory namespace.
92///
93/// Each function returns a [`WindowSpec`] compiled to the exact form the
94/// server's `WindowEngine.parseSpec` accepts.
95pub mod windows {
96    use super::WindowSpec;
97
98    /// Non-overlapping fixed windows: `tumbling("60s")`.
99    pub fn tumbling(size: &str) -> WindowSpec {
100        require_nonblank("size", size);
101        WindowSpec::new(format!("tumbling({size})"))
102    }
103
104    /// Overlapping windows: `sliding("10m", "1m")` = size, slide.
105    pub fn sliding(size: &str, slide: &str) -> WindowSpec {
106        require_nonblank("size", size);
107        require_nonblank("slide", slide);
108        WindowSpec::new(format!("sliding({size},{slide})"))
109    }
110
111    /// Inactivity-bounded windows: `session("30s")`.
112    pub fn session(timeout: &str) -> WindowSpec {
113        require_nonblank("timeout", timeout);
114        WindowSpec::new(format!("session({timeout})"))
115    }
116
117    /// Single unbounded window. Use for global aggregates.
118    pub fn global() -> WindowSpec {
119        WindowSpec::new("global")
120    }
121
122    /// Event-count tumbling: closes after `n` events. `count(100)`.
123    pub fn count(n: u64) -> WindowSpec {
124        if n == 0 {
125            panic!("count window size must be positive, got 0");
126        }
127        WindowSpec::new(format!("count({n})"))
128    }
129
130    /// Event-count sliding: `count_sliding(100, 10)` = window, slide.
131    pub fn count_sliding(size: u64, slide: u64) -> WindowSpec {
132        if size == 0 || slide == 0 {
133            panic!("count_sliding requires positive size and slide, got {size}, {slide}");
134        }
135        WindowSpec::new(format!("count_sliding({size},{slide})"))
136    }
137
138    fn require_nonblank(name: &str, value: &str) {
139        if value.trim().is_empty() {
140            panic!("{name} must be a non-empty string");
141        }
142    }
143}
144
145// ---------------------------------------------------------------------------
146// Aggregators
147// ---------------------------------------------------------------------------
148
149/// Aggregator factory namespace.
150///
151/// Each function returns the string template the server's `Aggregators.parse`
152/// accepts inside `window.aggregations` (e.g. `"avg(temperature)"`).
153pub mod aggs {
154    /// Event count — no field required.
155    pub fn count() -> String {
156        "count()".into()
157    }
158
159    /// Sum of a numeric field: `aggs::sum("amount")`.
160    pub fn sum(field: &str) -> String {
161        require_nonblank("field", field);
162        format!("sum({field})")
163    }
164
165    /// Average of a numeric field.
166    pub fn avg(field: &str) -> String {
167        require_nonblank("field", field);
168        format!("avg({field})")
169    }
170
171    /// Minimum value of a numeric field.
172    pub fn min(field: &str) -> String {
173        require_nonblank("field", field);
174        format!("min({field})")
175    }
176
177    /// Maximum value of a numeric field.
178    pub fn max(field: &str) -> String {
179        require_nonblank("field", field);
180        format!("max({field})")
181    }
182
183    /// Collect every value of `field` into a list.
184    pub fn collect_list(field: &str) -> String {
185        require_nonblank("field", field);
186        format!("collect_list({field})")
187    }
188
189    /// Cardinality of distinct values of `field`.
190    pub fn distinct_count(field: &str) -> String {
191        require_nonblank("field", field);
192        format!("distinct_count({field})")
193    }
194
195    fn require_nonblank(name: &str, value: &str) {
196        if value.trim().is_empty() {
197            panic!("{name} must be a non-empty string");
198        }
199    }
200}
201
202// ---------------------------------------------------------------------------
203// Option carriers
204// ---------------------------------------------------------------------------
205
206/// Options for [`StreamBuilder::map`].
207#[derive(Debug, Clone, Default)]
208pub struct MapOptions {
209    /// Output-field-name → source-expression-string mapping.
210    pub fields: Option<BTreeMap<String, String>>,
211    /// Tag the output event with a `type` field.
212    pub target_type: Option<String>,
213}
214
215/// Options for [`StreamBuilder::window`].
216#[derive(Debug, Clone, Default)]
217pub struct WindowOptions {
218    /// Map of output-field → aggregator-string (use [`aggs`] for the right-hand side).
219    pub aggregations: Option<BTreeMap<String, String>>,
220    /// Override for where window results go.
221    pub output_topic: Option<String>,
222    /// Server-side trigger config (passed through opaquely).
223    pub trigger: Option<Value>,
224}
225
226/// One branch of [`StreamBuilder::branch`].
227#[derive(Debug, Clone)]
228pub struct BranchSpec {
229    pub condition: String,
230    pub topic: String,
231}
232
233impl BranchSpec {
234    pub fn new(condition: impl Into<String>, topic: impl Into<String>) -> Self {
235        Self {
236            condition: condition.into(),
237            topic: topic.into(),
238        }
239    }
240}
241
242/// Options for [`StreamBuilder::enrich_async`].
243#[derive(Debug, Clone, Default)]
244pub struct EnrichAsyncOptions {
245    pub url: String,
246    pub parallelism: Option<u32>,
247    pub queue_size: Option<u32>,
248    pub timeout_ms: Option<u32>,
249    pub max_retries: Option<u32>,
250    pub retry_backoff_ms: Option<u32>,
251    /// Must be `"PRESERVE_INPUT"` or `"UNORDERED"`.
252    pub ordering: Option<String>,
253    /// Must be `"EMIT_ERROR"`, `"DROP"`, or `"PASS_THROUGH"`.
254    pub on_failure: Option<String>,
255}
256
257/// Options for [`StreamBuilder::cep`].
258#[derive(Debug, Clone, Default)]
259pub struct CepOptions {
260    pub within: Option<String>,
261    pub name: Option<String>,
262}
263
264/// Options for [`StreamBuilder::broadcast_join`].
265#[derive(Debug, Clone, Default)]
266pub struct BroadcastJoinOptions {
267    pub join_key_field: String,
268    pub streaming_topic: Option<String>,
269    pub name: Option<String>,
270    pub max_bytes: Option<i64>,
271    /// Must be `"cdc"`, `"periodic"`, or `"explicit"`.
272    pub refresh_mode: Option<String>,
273    pub interval_millis: Option<u32>,
274}
275
276/// Options for [`StreamBuilder::cdc_join`].
277#[derive(Debug, Clone, Default)]
278pub struct CdcJoinOptions {
279    pub source: String,
280    pub join_key: Option<String>,
281    pub table: Option<String>,
282    pub state_backend: Option<String>,
283}
284
285/// B-109 — options for [`StreamBuilder::map_llm`]. `output_field` is required.
286#[derive(Debug, Clone, Default)]
287pub struct MapLlmOptions {
288    pub output_field: String,
289    pub model: Option<String>,
290    pub temperature: Option<f64>,
291    pub max_tokens: Option<u32>,
292    pub parallelism: Option<u32>,
293    /// Must be `"PRESERVE_INPUT"` or `"UNORDERED"`.
294    pub ordering: Option<String>,
295    /// Must be `"EMIT_ERROR"`, `"DROP"`, or `"PASS_THROUGH"`.
296    pub on_failure: Option<String>,
297    pub max_calls_per_sec: Option<u32>,
298}
299
300/// B-109 — options for [`StreamBuilder::extract`]. `instruction` + `schema` required.
301#[derive(Debug, Clone, Default)]
302pub struct ExtractOptions {
303    pub instruction: String,
304    pub schema: BTreeMap<String, String>,
305    pub model: Option<String>,
306    pub temperature: Option<f64>,
307    pub max_tokens: Option<u32>,
308    pub on_failure: Option<String>,
309}
310
311/// B-109 Phase 2 — options for [`StreamBuilder::mcp_call`].
312#[derive(Debug, Clone, Default)]
313pub struct McpCallOptions {
314    pub args: Option<BTreeMap<String, Value>>,
315    pub output_field: Option<String>,
316    pub parallelism: Option<u32>,
317    pub ordering: Option<String>,
318    pub on_failure: Option<String>,
319}
320
321/// B-112 — options for [`StreamBuilder::ml_predict`]. `model`, `input_fields`
322/// and `output_field` are required.
323#[derive(Debug, Clone, Default)]
324pub struct MlPredictOptions {
325    /// Registered model name (upload first via `client.models().upload(...)`).
326    pub model: String,
327    /// Feature names pulled from the event, in the model's input order.
328    /// Dotted paths (`customer.tier`) resolve through nested objects.
329    pub input_fields: Vec<String>,
330    /// Event field the prediction object is written to.
331    pub output_field: String,
332    pub parallelism: Option<u32>,
333    /// Must be `"PRESERVE_INPUT"` or `"UNORDERED"`.
334    pub ordering: Option<String>,
335    /// Must be `"EMIT_ERROR"`, `"DROP"`, or `"PASS_THROUGH"`.
336    pub on_failure: Option<String>,
337}
338
339/// B-110 — options for [`StreamBuilder::wasm`]. `module` is required.
340#[derive(Debug, Clone, Default)]
341pub struct WasmOptions {
342    /// Registered module name (upload first via `client.wasm().upload(...)`).
343    pub module: String,
344    pub parallelism: Option<u32>,
345    /// Must be `"PRESERVE_INPUT"` or `"UNORDERED"`.
346    pub ordering: Option<String>,
347    /// Must be `"EMIT_ERROR"`, `"DROP"`, or `"PASS_THROUGH"`.
348    pub on_failure: Option<String>,
349}
350
351// ---------------------------------------------------------------------------
352// StreamBuilder
353// ---------------------------------------------------------------------------
354
355/// Fluent builder for a Pulse streaming pipeline.
356///
357/// Chain operator methods, then call [`build`](Self::build) (or pass to
358/// [`StreamsResource::deploy`]).
359///
360/// All operator methods take `&mut self` and return `Self` so calls chain
361/// naturally. Methods that validate their inputs panic on obviously-bad
362/// arguments (blank required fields, non-positive counts, unknown enum
363/// values) so bugs are caught at call site, not after a 400 round-trip.
364#[derive(Debug, Clone, Default)]
365pub struct StreamBuilder {
366    name: Option<String>,
367    description: Option<String>,
368    agent_label: Option<String>,
369    input_topic: Option<String>,
370    source_engine: Option<String>,
371    source_config: Map<String, Value>,
372    source_label: Option<String>,
373    output_topic: Option<String>,
374    sink_channel: Option<String>,
375    sink_config: Map<String, Value>,
376    sink_label: Option<String>,
377    operators: Vec<Map<String, Value>>,
378}
379
380impl StreamBuilder {
381    /// Builder with the given pipeline name preset.
382    pub fn new(name: impl Into<String>) -> Self {
383        let name = name.into();
384        require_nonblank("name", &name);
385        Self {
386            name: Some(name),
387            ..Self::default()
388        }
389    }
390
391    /// Builder with no preset name. Use [`named`](Self::named) or pass the
392    /// name to [`build_with_name`](Self::build_with_name).
393    pub fn anonymous() -> Self {
394        Self::default()
395    }
396
397    // ------------------------------------------------------------------
398    // Source
399    // ------------------------------------------------------------------
400
401    /// Sets the input topic. Source engine defaults to `"kafka"`.
402    pub fn from_topic(mut self, topic: impl Into<String>) -> Self {
403        let topic = topic.into();
404        require_nonblank("topic", &topic);
405        self.input_topic = Some(topic);
406        self.source_engine = Some("kafka".into());
407        self
408    }
409
410    /// Sets the input topic + source engine.
411    pub fn from_topic_with_engine(
412        mut self,
413        topic: impl Into<String>,
414        engine: impl Into<String>,
415    ) -> Self {
416        let topic = topic.into();
417        let engine = engine.into();
418        require_nonblank("topic", &topic);
419        require_nonblank("engine", &engine);
420        self.input_topic = Some(topic);
421        self.source_engine = Some(engine);
422        self
423    }
424
425    /// Merges extra config into the source node's `config` map.
426    pub fn with_source_config(mut self, key: impl Into<String>, value: Value) -> Self {
427        self.source_config.insert(key.into(), value);
428        self
429    }
430
431    /// Sets the display label for the source node.
432    pub fn with_source_label(mut self, label: impl Into<String>) -> Self {
433        self.source_label = Some(label.into());
434        self
435    }
436
437    // ------------------------------------------------------------------
438    // Operators
439    // ------------------------------------------------------------------
440
441    /// Filter operator. `condition` is a CEL-like expression string.
442    pub fn filter(mut self, condition: impl Into<String>) -> Self {
443        let condition = condition.into();
444        require_nonblank("condition", &condition);
445        let mut op = Map::new();
446        op.insert("type".into(), Value::String("filter".into()));
447        op.insert("condition".into(), Value::String(condition));
448        self.operators.push(op);
449        self
450    }
451
452    /// Map operator. At least one of `options.fields` / `options.target_type` is required.
453    pub fn map(mut self, options: MapOptions) -> Self {
454        if options.fields.is_none() && options.target_type.is_none() {
455            panic!("map operator does nothing — provide `fields` or `target_type`");
456        }
457        let mut op = Map::new();
458        op.insert("type".into(), Value::String("map".into()));
459        if let Some(fields) = options.fields {
460            let mut m = Map::new();
461            for (k, v) in fields {
462                m.insert(k, Value::String(v));
463            }
464            op.insert("fields".into(), Value::Object(m));
465        }
466        if let Some(t) = options.target_type {
467            op.insert("targetType".into(), Value::String(t));
468        }
469        self.operators.push(op);
470        self
471    }
472
473    /// Flat-map: explode an array-valued field into one event per element.
474    pub fn flat_map(mut self, split_field: impl Into<String>) -> Self {
475        let split_field = split_field.into();
476        require_nonblank("split_field", &split_field);
477        let mut op = Map::new();
478        op.insert("type".into(), Value::String("flatMap".into()));
479        op.insert("splitField".into(), Value::String(split_field));
480        self.operators.push(op);
481        self
482    }
483
484    /// Group the stream by a top-level field value. Required before stateful ops.
485    pub fn key_by(mut self, field: impl Into<String>) -> Self {
486        let field = field.into();
487        require_nonblank("field", &field);
488        let mut op = Map::new();
489        op.insert("type".into(), Value::String("keyBy".into()));
490        op.insert("field".into(), Value::String(field));
491        self.operators.push(op);
492        self
493    }
494
495    /// Window operator with no extra options.
496    pub fn window(self, spec: WindowSpec) -> Self {
497        self.window_full(spec, WindowOptions::default())
498    }
499
500    /// Window operator with aggregations only.
501    pub fn window_with_aggs(
502        self,
503        spec: WindowSpec,
504        aggregations: BTreeMap<String, String>,
505    ) -> Self {
506        self.window_full(
507            spec,
508            WindowOptions {
509                aggregations: Some(aggregations),
510                ..Default::default()
511            },
512        )
513    }
514
515    /// Window operator with the full option set.
516    pub fn window_full(mut self, spec: WindowSpec, options: WindowOptions) -> Self {
517        let mut op = Map::new();
518        op.insert("type".into(), Value::String("window".into()));
519        op.insert("spec".into(), Value::String(spec.spec.clone()));
520        if let Some(aggs_map) = options.aggregations {
521            let mut m = Map::new();
522            for (k, v) in aggs_map {
523                m.insert(k, Value::String(v));
524            }
525            op.insert("aggregations".into(), Value::Object(m));
526        }
527        if let Some(out) = options.output_topic {
528            op.insert("outputTopic".into(), Value::String(out));
529        }
530        if let Some(trig) = options.trigger {
531            op.insert("trigger".into(), trig);
532        }
533        self.operators.push(op);
534        self
535    }
536
537    /// Window operator with a raw spec string. Useful when you've already
538    /// validated the spec against `WindowEngine.parseSpec`.
539    pub fn window_from_str(mut self, spec: &str, options: WindowOptions) -> Self {
540        require_nonblank("spec", spec);
541        self = self.window_full(WindowSpec::new(spec), options);
542        self
543    }
544
545    /// Branch operator: route events to different topics by condition.
546    pub fn branch(mut self, branches: Vec<BranchSpec>) -> Self {
547        if branches.is_empty() {
548            panic!("branch operator requires at least one branch");
549        }
550        let mut normalised = Vec::with_capacity(branches.len());
551        for (i, b) in branches.iter().enumerate() {
552            if b.condition.trim().is_empty() {
553                panic!("branch[{i}] requires a non-empty `condition`");
554            }
555            if b.topic.trim().is_empty() {
556                panic!("branch[{i}] requires a non-empty `topic`");
557            }
558            normalised.push(json!({
559                "condition": b.condition,
560                "topic": b.topic,
561            }));
562        }
563        let mut op = Map::new();
564        op.insert("type".into(), Value::String("branch".into()));
565        op.insert("branches".into(), Value::Array(normalised));
566        self.operators.push(op);
567        self
568    }
569
570    /// Synchronous enrichment: join the stream against a state-store topic.
571    pub fn enrich(mut self, lookup_topic: impl Into<String>, key_field: impl Into<String>) -> Self {
572        let lookup_topic = lookup_topic.into();
573        let key_field = key_field.into();
574        require_nonblank("lookup_topic", &lookup_topic);
575        require_nonblank("key_field", &key_field);
576        let mut op = Map::new();
577        op.insert("type".into(), Value::String("enrich".into()));
578        op.insert("lookupTopic".into(), Value::String(lookup_topic));
579        op.insert("keyField".into(), Value::String(key_field));
580        self.operators.push(op);
581        self
582    }
583
584    /// Asynchronous HTTP enrichment. `url` supports `{field}` placeholders.
585    pub fn enrich_async(mut self, options: EnrichAsyncOptions) -> Self {
586        require_nonblank("url", &options.url);
587        if let Some(ref o) = options.ordering {
588            if o != "PRESERVE_INPUT" && o != "UNORDERED" {
589                panic!("ordering must be PRESERVE_INPUT or UNORDERED, got {o:?}");
590            }
591        }
592        if let Some(ref f) = options.on_failure {
593            if f != "EMIT_ERROR" && f != "DROP" && f != "PASS_THROUGH" {
594                panic!("on_failure must be EMIT_ERROR, DROP, or PASS_THROUGH, got {f:?}");
595            }
596        }
597        let mut op = Map::new();
598        op.insert("type".into(), Value::String("enrichAsync".into()));
599        op.insert("url".into(), Value::String(options.url));
600        if let Some(v) = options.parallelism {
601            op.insert("parallelism".into(), Value::Number(v.into()));
602        }
603        if let Some(v) = options.queue_size {
604            op.insert("queueSize".into(), Value::Number(v.into()));
605        }
606        if let Some(v) = options.timeout_ms {
607            op.insert("timeoutMs".into(), Value::Number(v.into()));
608        }
609        if let Some(v) = options.max_retries {
610            op.insert("maxRetries".into(), Value::Number(v.into()));
611        }
612        if let Some(v) = options.retry_backoff_ms {
613            op.insert("retryBackoffMs".into(), Value::Number(v.into()));
614        }
615        if let Some(o) = options.ordering {
616            op.insert("ordering".into(), Value::String(o));
617        }
618        if let Some(f) = options.on_failure {
619            op.insert("onFailure".into(), Value::String(f));
620        }
621        self.operators.push(op);
622        self
623    }
624
625    /// Complex Event Processing: match a sequence of conditions.
626    pub fn cep(mut self, sequence: Vec<Value>, options: CepOptions) -> Self {
627        if sequence.is_empty() {
628            panic!("cep operator requires a non-empty sequence");
629        }
630        let mut op = Map::new();
631        op.insert("type".into(), Value::String("cep".into()));
632        op.insert("sequence".into(), Value::Array(sequence));
633        if let Some(w) = options.within {
634            op.insert("within".into(), Value::String(w));
635        }
636        if let Some(n) = options.name {
637            op.insert("name".into(), Value::String(n));
638        }
639        self.operators.push(op);
640        self
641    }
642
643    /// B-109 — enrich each event with an LLM completion. `prompt` supports
644    /// `{field}` placeholders (and `{__payload__}`) substituted from the event
645    /// server-side; the completion lands on the event under `output_field`.
646    pub fn map_llm(mut self, prompt: impl Into<String>, options: MapLlmOptions) -> Self {
647        let prompt = prompt.into();
648        require_nonblank("prompt", &prompt);
649        require_nonblank("output_field", &options.output_field);
650        if let Some(ref o) = options.ordering {
651            if o != "PRESERVE_INPUT" && o != "UNORDERED" {
652                panic!("ordering must be PRESERVE_INPUT or UNORDERED, got {o:?}");
653            }
654        }
655        check_failure(&options.on_failure);
656        let mut op = Map::new();
657        op.insert("type".into(), Value::String("mapLlm".into()));
658        op.insert("prompt".into(), Value::String(prompt));
659        op.insert("outputField".into(), Value::String(options.output_field));
660        if let Some(m) = options.model {
661            op.insert("model".into(), Value::String(m));
662        }
663        if let Some(t) = options.temperature {
664            op.insert("temperature".into(), json!(t));
665        }
666        if let Some(n) = options.max_tokens {
667            op.insert("maxTokens".into(), Value::Number(n.into()));
668        }
669        if let Some(n) = options.parallelism {
670            op.insert("parallelism".into(), Value::Number(n.into()));
671        }
672        if let Some(o) = options.ordering {
673            op.insert("ordering".into(), Value::String(o));
674        }
675        if let Some(f) = options.on_failure {
676            op.insert("onFailure".into(), Value::String(f));
677        }
678        if let Some(n) = options.max_calls_per_sec {
679            op.insert("maxCallsPerSec".into(), Value::Number(n.into()));
680        }
681        self.operators.push(op);
682        self
683    }
684
685    /// B-109 — LLM → typed structured fields merged into the event. The LLM is
686    /// asked for a JSON object keyed by `options.schema`'s fields; missing /
687    /// malformed fields become null server-side.
688    pub fn extract(mut self, options: ExtractOptions) -> Self {
689        require_nonblank("instruction", &options.instruction);
690        if options.schema.is_empty() {
691            panic!("extract operator requires a non-empty schema");
692        }
693        check_failure(&options.on_failure);
694        let mut schema = Map::new();
695        for (k, v) in options.schema {
696            schema.insert(k, Value::String(v));
697        }
698        let mut op = Map::new();
699        op.insert("type".into(), Value::String("extract".into()));
700        op.insert("instruction".into(), Value::String(options.instruction));
701        op.insert("schema".into(), Value::Object(schema));
702        if let Some(m) = options.model {
703            op.insert("model".into(), Value::String(m));
704        }
705        if let Some(t) = options.temperature {
706            op.insert("temperature".into(), json!(t));
707        }
708        if let Some(n) = options.max_tokens {
709            op.insert("maxTokens".into(), Value::Number(n.into()));
710        }
711        if let Some(f) = options.on_failure {
712            op.insert("onFailure".into(), Value::String(f));
713        }
714        self.operators.push(op);
715        self
716    }
717
718    /// B-109 Phase 2 — invoke an MCP tool per event. `options.args` string
719    /// values support `{field}` substitution. On success the tool output is
720    /// written to `options.output_field` (omit for a fire-and-forget effect).
721    pub fn mcp_call(mut self, tool: impl Into<String>, options: McpCallOptions) -> Self {
722        let tool = tool.into();
723        require_nonblank("tool", &tool);
724        if let Some(ref o) = options.ordering {
725            if o != "PRESERVE_INPUT" && o != "UNORDERED" {
726                panic!("ordering must be PRESERVE_INPUT or UNORDERED, got {o:?}");
727            }
728        }
729        check_failure(&options.on_failure);
730        let mut op = Map::new();
731        op.insert("type".into(), Value::String("mcpCall".into()));
732        op.insert("tool".into(), Value::String(tool));
733        if let Some(args) = options.args {
734            let mut m = Map::new();
735            for (k, v) in args {
736                m.insert(k, v);
737            }
738            op.insert("args".into(), Value::Object(m));
739        }
740        if let Some(f) = options.output_field {
741            op.insert("outputField".into(), Value::String(f));
742        }
743        if let Some(n) = options.parallelism {
744            op.insert("parallelism".into(), Value::Number(n.into()));
745        }
746        if let Some(o) = options.ordering {
747            op.insert("ordering".into(), Value::String(o));
748        }
749        if let Some(f) = options.on_failure {
750            op.insert("onFailure".into(), Value::String(f));
751        }
752        self.operators.push(op);
753        self
754    }
755
756    /// B-112 — score each event with an embedded ML model. The uploaded ONNX
757    /// model runs in-process on the Pulse engine (no model-server hop): the
758    /// named `options.input_fields` are pulled from the event payload, fed to
759    /// the model, and the model's output is written as a nested object under
760    /// `options.output_field` (so downstream operators can branch on it, e.g.
761    /// `.filter("prediction.fraud_score > 0.8")`).
762    ///
763    /// Upload the model first with [`ModelsResource::upload`](crate::ModelsResource::upload).
764    pub fn ml_predict(mut self, options: MlPredictOptions) -> Self {
765        require_nonblank("model", &options.model);
766        require_nonblank("output_field", &options.output_field);
767        if options.input_fields.is_empty()
768            || options.input_fields.iter().any(|f| f.trim().is_empty())
769        {
770            panic!("input_fields must be a non-empty list of non-blank strings");
771        }
772        if let Some(ref o) = options.ordering {
773            if o != "PRESERVE_INPUT" && o != "UNORDERED" {
774                panic!("ordering must be PRESERVE_INPUT or UNORDERED, got {o:?}");
775            }
776        }
777        check_failure(&options.on_failure);
778        let mut op = Map::new();
779        op.insert("type".into(), Value::String("mlPredict".into()));
780        op.insert("model".into(), Value::String(options.model));
781        op.insert(
782            "inputFields".into(),
783            Value::Array(
784                options
785                    .input_fields
786                    .into_iter()
787                    .map(Value::String)
788                    .collect(),
789            ),
790        );
791        op.insert("outputField".into(), Value::String(options.output_field));
792        if let Some(n) = options.parallelism {
793            op.insert("parallelism".into(), Value::Number(n.into()));
794        }
795        if let Some(o) = options.ordering {
796            op.insert("ordering".into(), Value::String(o));
797        }
798        if let Some(f) = options.on_failure {
799            op.insert("onFailure".into(), Value::String(f));
800        }
801        self.operators.push(op);
802        self
803    }
804
805    /// B-110 — run a sandboxed WASM module over each event. The uploaded module
806    /// (see [`WasmResource::upload`](crate::WasmResource::upload)) receives the
807    /// event payload bytes and returns the new payload (transform / map) or
808    /// drops the event (filter), running in pure-Java Chicory on the engine —
809    /// no host syscalls, bounded linear memory. Any `wasm32` toolchain (Rust,
810    /// TinyGo, AssemblyScript, C) can author a module against the alloc/process
811    /// ABI.
812    pub fn wasm(mut self, options: WasmOptions) -> Self {
813        require_nonblank("module", &options.module);
814        if let Some(ref o) = options.ordering {
815            if o != "PRESERVE_INPUT" && o != "UNORDERED" {
816                panic!("ordering must be PRESERVE_INPUT or UNORDERED, got {o:?}");
817            }
818        }
819        check_failure(&options.on_failure);
820        let mut op = Map::new();
821        op.insert("type".into(), Value::String("wasm".into()));
822        op.insert("module".into(), Value::String(options.module));
823        if let Some(n) = options.parallelism {
824            op.insert("parallelism".into(), Value::Number(n.into()));
825        }
826        if let Some(o) = options.ordering {
827            op.insert("ordering".into(), Value::String(o));
828        }
829        if let Some(f) = options.on_failure {
830            op.insert("onFailure".into(), Value::String(f));
831        }
832        self.operators.push(op);
833        self
834    }
835
836    /// Broadcast join: enrich the stream against a fully-replicated table.
837    pub fn broadcast_join(mut self, options: BroadcastJoinOptions) -> Self {
838        require_nonblank("join_key_field", &options.join_key_field);
839        if let Some(ref m) = options.refresh_mode {
840            if m != "cdc" && m != "periodic" && m != "explicit" {
841                panic!("refresh_mode must be cdc, periodic, or explicit, got {m:?}");
842            }
843        }
844        let mut op = Map::new();
845        op.insert("type".into(), Value::String("broadcastJoin".into()));
846        op.insert("joinKeyField".into(), Value::String(options.join_key_field));
847        if let Some(t) = options.streaming_topic {
848            op.insert("streamingTopic".into(), Value::String(t));
849        }
850        if let Some(n) = options.name {
851            op.insert("name".into(), Value::String(n));
852        }
853        if let Some(b) = options.max_bytes {
854            op.insert("maxBytes".into(), Value::Number(b.into()));
855        }
856        if let Some(m) = options.refresh_mode {
857            op.insert("refreshMode".into(), Value::String(m));
858        }
859        if let Some(i) = options.interval_millis {
860            op.insert("intervalMillis".into(), Value::Number(i.into()));
861        }
862        self.operators.push(op);
863        self
864    }
865
866    /// CDC join: stream-table join against a CDC-fed state table.
867    pub fn cdc_join(mut self, options: CdcJoinOptions) -> Self {
868        require_nonblank("source", &options.source);
869        let mut op = Map::new();
870        op.insert("type".into(), Value::String("cdcJoin".into()));
871        op.insert("source".into(), Value::String(options.source));
872        if let Some(k) = options.join_key {
873            op.insert("joinKey".into(), Value::String(k));
874        }
875        if let Some(t) = options.table {
876            op.insert("table".into(), Value::String(t));
877        }
878        if let Some(b) = options.state_backend {
879            op.insert("stateBackend".into(), Value::String(b));
880        }
881        self.operators.push(op);
882        self
883    }
884
885    // ------------------------------------------------------------------
886    // Sink
887    // ------------------------------------------------------------------
888
889    /// Sets the output topic only. No sink node is emitted.
890    pub fn to_topic(mut self, topic: impl Into<String>) -> Self {
891        let topic = topic.into();
892        require_nonblank("topic", &topic);
893        self.output_topic = Some(topic);
894        self.sink_channel = None;
895        self
896    }
897
898    /// Sets the output topic + sink channel (emits a sink node).
899    pub fn to_topic_with_channel(
900        mut self,
901        topic: impl Into<String>,
902        channel: impl Into<String>,
903    ) -> Self {
904        let topic = topic.into();
905        let channel = channel.into();
906        require_nonblank("topic", &topic);
907        require_nonblank("channel", &channel);
908        self.output_topic = Some(topic);
909        self.sink_channel = Some(channel);
910        self
911    }
912
913    /// Terminate the stream in a connector sink (Segment, Kafka, Postgres, …) —
914    /// an ergonomic, connector-first alias for
915    /// [`to_topic_with_channel`](Self::to_topic_with_channel) using an
916    /// intermediate `<connector_type>-sink-out` topic. Chain
917    /// [`with_sink_config`](Self::with_sink_config) for the connector config.
918    /// `connector_type` is a subType from `client.connectors()`; bridged
919    /// connectors require the enterprise bridge JAR on the server.
920    pub fn to_connector(self, connector_type: impl Into<String>) -> Self {
921        let ct = connector_type.into();
922        require_nonblank("connector_type", &ct);
923        let topic = format!("{ct}-sink-out");
924        self.to_topic_with_channel(topic, ct)
925    }
926
927    /// Merges extra config into the sink node's `config` map.
928    pub fn with_sink_config(mut self, key: impl Into<String>, value: Value) -> Self {
929        self.sink_config.insert(key.into(), value);
930        self
931    }
932
933    /// Sets the display label for the sink node.
934    pub fn with_sink_label(mut self, label: impl Into<String>) -> Self {
935        self.sink_label = Some(label.into());
936        self
937    }
938
939    /// Terminate the stream in the agent's state store (queryable via B-106 IQ).
940    pub fn to_state(mut self) -> Self {
941        self.output_topic = None;
942        self.sink_channel = None;
943        self.sink_config = Map::new();
944        self.sink_label = None;
945        self
946    }
947
948    // ------------------------------------------------------------------
949    // Metadata
950    // ------------------------------------------------------------------
951
952    /// Sets / overrides the pipeline name.
953    pub fn named(mut self, name: impl Into<String>) -> Self {
954        let name = name.into();
955        require_nonblank("name", &name);
956        self.name = Some(name);
957        self
958    }
959
960    /// Sets the pipeline description.
961    pub fn described_as(mut self, description: impl Into<String>) -> Self {
962        self.description = Some(description.into());
963        self
964    }
965
966    /// Sets the display label for the streaming agent node.
967    pub fn with_agent_label(mut self, label: impl Into<String>) -> Self {
968        let label = label.into();
969        require_nonblank("label", &label);
970        self.agent_label = Some(label);
971        self
972    }
973
974    // ------------------------------------------------------------------
975    // Compilation
976    // ------------------------------------------------------------------
977
978    /// Returns a read-only view of the recorded operator chain.
979    pub fn operators(&self) -> &[Map<String, Value>] {
980        &self.operators
981    }
982
983    /// Compile the chain into a Pulse pipeline dict ready for POST.
984    pub fn build(&self) -> Result<Value, PulseError> {
985        self.build_inner(None)
986    }
987
988    /// Same as [`build`](Self::build) but overrides the pipeline name.
989    pub fn build_with_name(&self, name: &str) -> Result<Value, PulseError> {
990        require_nonblank("name", name);
991        self.build_inner(Some(name.to_string()))
992    }
993
994    fn build_inner(&self, override_name: Option<String>) -> Result<Value, PulseError> {
995        let pipeline_name = override_name.or_else(|| self.name.clone()).ok_or_else(|| {
996            PulseError::InvalidConfig(
997                "pipeline name required — pass to StreamBuilder::new or build_with_name".into(),
998            )
999        })?;
1000        let input_topic = self.input_topic.as_ref().ok_or_else(|| {
1001            PulseError::InvalidConfig("no source — call .from_topic(...) before build()".into())
1002        })?;
1003        if self.operators.is_empty() {
1004            return Err(PulseError::InvalidConfig(
1005                "no operators — chain at least one of .filter/.map/.key_by/... before build()"
1006                    .into(),
1007            ));
1008        }
1009
1010        let source_engine = self.source_engine.as_deref().unwrap_or("kafka");
1011
1012        let mut nodes: Vec<Value> = Vec::with_capacity(3);
1013
1014        // Source node
1015        let mut src_config = Map::new();
1016        src_config.insert("engine".into(), Value::String(source_engine.to_string()));
1017        src_config.insert("inputTopic".into(), Value::String(input_topic.clone()));
1018        for (k, v) in &self.source_config {
1019            src_config.insert(k.clone(), v.clone());
1020        }
1021        let src_label = self
1022            .source_label
1023            .clone()
1024            .unwrap_or_else(|| format!("{source_engine} source"));
1025        nodes.push(json!({
1026            "type": "source",
1027            "label": src_label,
1028            "config": Value::Object(src_config),
1029        }));
1030
1031        // Agent node
1032        let mut agent_config = Map::new();
1033        agent_config.insert("engine".into(), Value::String("streaming".into()));
1034        agent_config.insert("inputTopic".into(), Value::String(input_topic.clone()));
1035        let ops_value: Vec<Value> = self
1036            .operators
1037            .iter()
1038            .map(|op| Value::Object(op.clone()))
1039            .collect();
1040        agent_config.insert("operators".into(), Value::Array(ops_value));
1041        if let Some(ref out) = self.output_topic {
1042            agent_config.insert("outputTopic".into(), Value::String(out.clone()));
1043        }
1044        let agent_label = self
1045            .agent_label
1046            .clone()
1047            .unwrap_or_else(|| pipeline_name.clone());
1048        nodes.push(json!({
1049            "type": "agent",
1050            "label": agent_label,
1051            "config": Value::Object(agent_config),
1052        }));
1053
1054        // Sink node — only when both output_topic AND sink_channel are set
1055        if let (Some(out), Some(ch)) = (self.output_topic.as_ref(), self.sink_channel.as_ref()) {
1056            let mut sink_conf = Map::new();
1057            sink_conf.insert("channel".into(), Value::String(ch.clone()));
1058            sink_conf.insert("inputTopic".into(), Value::String(out.clone()));
1059            for (k, v) in &self.sink_config {
1060                sink_conf.insert(k.clone(), v.clone());
1061            }
1062            let sink_label = self
1063                .sink_label
1064                .clone()
1065                .unwrap_or_else(|| format!("{ch} sink"));
1066            nodes.push(json!({
1067                "type": "sink",
1068                "label": sink_label,
1069                "config": Value::Object(sink_conf),
1070            }));
1071        }
1072
1073        let mut pipeline = Map::new();
1074        pipeline.insert("name".into(), Value::String(pipeline_name));
1075        pipeline.insert("nodes".into(), Value::Array(nodes));
1076        if let Some(ref desc) = self.description {
1077            pipeline.insert("description".into(), Value::String(desc.clone()));
1078        }
1079        Ok(Value::Object(pipeline))
1080    }
1081}
1082
1083// ---------------------------------------------------------------------------
1084// StreamsResource — the client.streams() accessor
1085// ---------------------------------------------------------------------------
1086
1087/// `client.streams()` — compile + deploy [`StreamBuilder`] pipelines.
1088///
1089/// Sugar over `client.pipelines().create()` — the compile happens client-side,
1090/// the deploy is the same POST.
1091pub struct StreamsResource<'c> {
1092    pub(crate) client: &'c PulseClient,
1093}
1094
1095impl<'c> StreamsResource<'c> {
1096    /// Compile the builder to a pipeline dict WITHOUT deploying.
1097    pub fn compile(&self, builder: &StreamBuilder) -> Result<Value, PulseError> {
1098        builder.build()
1099    }
1100
1101    /// Compile with a name override WITHOUT deploying.
1102    pub fn compile_with_name(
1103        &self,
1104        builder: &StreamBuilder,
1105        name: &str,
1106    ) -> Result<Value, PulseError> {
1107        builder.build_with_name(name)
1108    }
1109
1110    /// Compile + POST to `/api/pulse/pipelines`. Returns the server response.
1111    pub async fn deploy(&self, builder: &StreamBuilder) -> Result<Value, PulseError> {
1112        let definition = builder.build()?;
1113        self.client
1114            .request(
1115                Method::POST,
1116                "/api/pulse/pipelines",
1117                Some(&definition),
1118                true,
1119            )
1120            .await
1121    }
1122
1123    /// Compile with a name override + POST to `/api/pulse/pipelines`.
1124    pub async fn deploy_with_name(
1125        &self,
1126        builder: &StreamBuilder,
1127        name: &str,
1128    ) -> Result<Value, PulseError> {
1129        let definition = builder.build_with_name(name)?;
1130        self.client
1131            .request(
1132                Method::POST,
1133                "/api/pulse/pipelines",
1134                Some(&definition),
1135                true,
1136            )
1137            .await
1138    }
1139}
1140
1141impl std::fmt::Debug for StreamsResource<'_> {
1142    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1143        f.debug_struct("StreamsResource").finish()
1144    }
1145}
1146
1147// ---------------------------------------------------------------------------
1148// Internal helpers
1149// ---------------------------------------------------------------------------
1150
1151fn require_nonblank(name: &str, value: &str) {
1152    if value.trim().is_empty() {
1153        panic!("{name} must be a non-empty string");
1154    }
1155}
1156
1157/// Panics if `on_failure` is set to an invalid value (B-109).
1158fn check_failure(on_failure: &Option<String>) {
1159    if let Some(f) = on_failure {
1160        if f != "EMIT_ERROR" && f != "DROP" && f != "PASS_THROUGH" {
1161            panic!("on_failure must be EMIT_ERROR, DROP, or PASS_THROUGH, got {f:?}");
1162        }
1163    }
1164}
pulse_client/streams.rs

pulse_client/
streams.rs
