🥮 mooncakes.io
    Sign in

    This version of the module is deprecated: Withdrawn: published under the wrong account. This project does not belong to JSJ222. Do not use this package.

    mooncep

    Bounded event-time complex event pattern detection for MoonBit

    cep
    events
    patterns
    event-time
    streaming
    Download zip
    Author
    Version
    0.1.0
    License
    MIT
    Repository
    Last updated
    11 hours ago
    Downloads
    3

    Dependencies

    #MoonCEP

    MoonCEP is an embeddable complex event pattern detector written in MoonBit. It recognizes named sequences in keyed, out-of-order business events and waits for caller-provided watermarks before confirming time-bounded absence.

    MoonCEP 面向需要跨多条事件识别业务过程的开发者：账号异常登录、订单超时未支付、 设备告警与恢复、用户行为漏斗。核心输入是已经发生的事件，输出是命名捕获及事件证据。 它不运行外部业务动作，也不要求部署流计算集群。

    #Install

    moon add JSJ222/mooncep@0.1.0

    Add the library to your consuming package's moon.pkg:

    import {
      "JSJ222/mooncep" @cep,
    }

    The library imports only moonbitlang/core. The module additionally depends on moonbitlang/x@0.4.49 for the file CLI. Tested with moonc 0.10.16 on wasm, wasm-gc, JavaScript and native; CI checks and builds all four targets.

    #Recognize a sequence

    fn main raise {
      let pattern = @cep.Pattern::new("login-signal", [
        @cep.Stage::new("failures", @cep.Kind("failed"), min=3, max=3),
        @cep.Stage::new("success", @cep.Kind("success")),
      ], 300000L, after_match=@cep.SkipPastLast)
      let engine = @cep.Engine::new([pattern])
      for (id, kind, time) in [
        ("f1", "failed", 1000L), ("f2", "failed", 2000L),
        ("f3", "failed", 3000L), ("s1", "success", 4000L),
      ] {
        ignore(engine.push(@cep.Event::new(id, "user/7", time, kind)))
      }
      let batch = engine.advance_watermark(4000L)
      assert_eq(batch.matches().length(), 1)
      println(@cep.match_to_json(batch.matches()[0]).stringify())
    }

    All timestamps use caller-chosen consistent integer units. Examples use milliseconds. A watermark W closes all event times <= W. Events at the same timestamp follow successful input occurrence order; an event at an already closed timestamp is rejected. finish() closes every supported deadline.

    #Capabilities and boundaries

    • Named stages, finite min..max repetitions, optional stages (min=0).
    • Strict, skip-next and skip-any contiguity; three explicit after-match policies.
    • Typed field predicates, Boolean combinations and earlier-capture comparisons.
    • Stage exclusion predicates and trailing absence through a closed deadline.
    • Key isolation, event-time heap, deterministic capture evidence and timeouts.
    • Atomic limits on buffered events, states, keys, selected outputs and work units.
    • Detached checkpoints, versioned JSON session journals, bounded replay and preview.
    • Incremental NDJSON framing, transactional stream ingestion, typed alert projections, partial-match diagnostics and bounded detection telemetry.

    Version 1 has finite patterns and synchronous, single-owner state. It does not provide distributed execution, unbounded repetition, pattern alternation/groups, greedy matching, event corrections/retractions, cross-key joins, arbitrary user callbacks, automatic watermarks, queue connectors or external-sink exactly-once delivery. It defines its own semantics and does not claim FlinkCEP compatibility. See semantics, API guide and resource/complexity model.

    #Run complete examples

    From a source checkout, run moon update, then:

    moon run --target native examples/security
moon run --target native examples/orders
moon run --target native examples/equipment
moon run --target native examples/funnel
moon run --target native --release examples/performance

    ExampleComplete input-to-output scenario
    SecurityThree failed logins followed by success for the same account produce captured evidence.
    OrdersA persisted session is restored; unpaid orders are confirmed after their deadline, while payment at the deadline cancels absence.
    EquipmentOut-of-order alarms, a rising field comparison and recovery produce a typed operational record.
    FunnelAn optional coupon stage and multiple purchases demonstrate preview, branching and result telemetry.

    The synthetic fixtures were created for this repository; no production data is included.

    #File CLI

    moon run --target native cmd/mooncep -- validate examples/files/program.json
moon run --target native cmd/mooncep -- run examples/files/program.json examples/files/events.ndjson
moon run --target native cmd/mooncep -- inspect SESSION.json
moon run --target native cmd/mooncep -- restore SESSION.json EVENTS.ndjson

    run emits JSON batches, configured alert records and final counters. restore emits batches from new commands and finalization; it does not replay old results to a sink. The CLI loads each file into memory and bounds it to 16 Mi UTF-16 units; library text documents and individual NDJSON lines have smaller bounds. See wire formats.

    On the affected Windows 0.10.16 toolchain, replace moon ... with & .\scripts\Invoke-Moon.ps1 .... The wrapper adapts extended drive paths for the installed official compiler and formatter and restores process environment settings. It does not change the installed tools. Other toolchain versions use normal commands.

    #Verify

    moon check --target all --deny-warn
moon build --target all --deny-warn
moon test --target all --deny-warn
moon fmt --check
moon info
git diff --exit-code
python scripts/verify_repository.py

    Tests include independent exhaustive subsequence enumeration, optional/repeated stages, skip policies, window equality, negative predicates, late data, rollback, exact Int64 values, Unicode chunk boundaries and session ownership.

    MIT licensed. This is an independent implementation with AI assistance, not a source-code port of Apache Flink. Conceptual references and dependency licensing are documented in sources. See contributing for maintenance expectations and CHANGELOG for release scope.

    CepError

    source
    pub(all) suberror CepError {
      Invalid(path~ : String, reason~ : String)
      Resource(resource~ : String, limit~ : Int)
      LateEvent(time~ : Int64, watermark~ : Int64)
      WatermarkRegression(previous~ : Int64, requested~ : Int64)
    } derive(
    Debug    )

    CepError::message

    source
    fn CepError::message(self : CepError) -> String

    CepError::to_repr

    source

    AfterMatch

    source
    pub(all) enum AfterMatch {
      NoSkip
      SkipToNextStart
      SkipPastLast
    } derive(Eq, 
    Debug    )

    Selection applied after a match has been emitted.

    AfterMatch::equal

    source
    fn AfterMatch::equal(AfterMatch, AfterMatch) -> Bool

    AfterMatch::not_equal

    source
    fn AfterMatch::not_equal(x : AfterMatch, y : AfterMatch) -> Bool

    AfterMatch::to_repr

    source

    AlertRecord

    source
    pub(all) struct AlertRecord {
      pattern : String
      key : String
      identity : String
      fields : Array[Field]
    }

    Batch

    source
    pub struct Batch {
      matches : Array[Match]
      timeouts : Array[Timeout]
      processed : Int
      work : Int
    }

    Batch::matches

    source
    fn Batch::matches(self : Batch) -> Array[Match]

    Batch::processed

    source
    fn Batch::processed(self : Batch) -> Int

    Batch::timeouts

    source
    fn Batch::timeouts(self : Batch) -> Array[Timeout]

    Batch::work

    source
    fn Batch::work(self : Batch) -> Int

    Branch

    source
    type Branch

    BranchingRisk

    source
    pub(all) enum BranchingRisk {
      FixedSequence
      BoundedRepetition
      CombinatorialSelection
    } derive(Eq, 
    Debug    )

    Structural classification, not a guarantee on runtime candidate count.

    BranchingRisk::equal

    source

    BranchingRisk::not_equal

    source
    fn BranchingRisk::not_equal(x : BranchingRisk, y : BranchingRisk) -> Bool

    BranchingRisk::to_repr

    source

    Capture

    source
    pub(all) struct Capture {
      name : String
      events : Array[Event]
    } derive(Eq, 
    Debug    )

    Events captured by one named positive stage, in their matching order.

    Capture::equal

    source
    fn Capture::equal(Capture, Capture) -> Bool

    Capture::not_equal

    source
    fn Capture::not_equal(x : Capture, y : Capture) -> Bool

    Capture::to_repr

    source

    Checkpoint

    source
    pub struct Checkpoint {
      state : Engine
    }

    An owned, in-process checkpoint. This is not a serialized durable format and does not coordinate external sources, result delivery or transactions.

    Checkpoint::active_states

    source
    fn Checkpoint::active_states(self : Checkpoint) -> Int

    Checkpoint::buffered

    source
    fn Checkpoint::buffered(self : Checkpoint) -> Int

    Checkpoint::restore

    source
    fn Checkpoint::restore(self : Checkpoint) -> Engine

    Each restore returns a new engine, independent of all earlier restores.

    Checkpoint::restore_with_limits

    source
    fn Checkpoint::restore_with_limits(self : Checkpoint, limits : Limits) -> Engine raise CepError

    Restores independently with validated replacement budgets.

    Checkpoint::watermark

    source
    fn Checkpoint::watermark(self : Checkpoint) -> Int64

    Column

    source
    pub(all) struct Column {
      name : String
      projection : Projection
    }

    CommandReader

    source
    pub struct CommandReader {
      partial : String
      line : Int
      closed : Bool
      max_line : Int
      max_commands : Int
    }

    Incremental UTF-16 text framing for newline-delimited stream commands. CRLF and blank lines are accepted. Parsing and reader state commit together.

    CommandReader::feed

    source
    fn CommandReader::feed(self : CommandReader, chunk : String) -> Array[StreamInput] raise CepError

    CommandReader::finish

    source

    CommandReader::fork

    source

    CommandReader::is_finished

    source
    fn CommandReader::is_finished(self : CommandReader) -> Bool

    CommandReader::line

    source
    fn CommandReader::line(self : CommandReader) -> Int

    CommandReader::new

    source
    fn CommandReader::new(max_line? : Int, max_commands? : Int) -> CommandReader raise CepError

    CommandReader::pending_units

    source
    fn CommandReader::pending_units(self : CommandReader) -> Int

    Comparison

    source
    pub(all) enum Comparison {
      Equal
      NotEqual
      Less
      LessEqual
      Greater
      GreaterEqual
    } derive(Eq, 
    Debug    )

    Comparison::equal

    source
    fn Comparison::equal(Comparison, Comparison) -> Bool

    Comparison::not_equal

    source
    fn Comparison::not_equal(x : Comparison, y : Comparison) -> Bool

    Comparison::to_repr

    source

    Condition

    source
    pub(all) enum Condition {
      Always
      Kind(String)
      Has(String)
      Compare(String, Comparison, Value)
      FieldCompare(String, Comparison, String)
      CaptureCompare(String, Comparison, String, String)
      CaptureCount(String, Comparison, Int)
      TextPrefix(String, String)
      TextContains(String, String)
      All(Array[Condition])
      Any(Array[Condition])
      Not(Condition)
    } derive(Eq, 
    Debug    )

    A pure condition over the current event and previously completed captures. Arrays are detached when a Stage or Pattern is constructed.

    Condition::equal

    source
    fn Condition::equal(Condition, Condition) -> Bool

    Condition::evaluate

    source
    fn Condition::evaluate(self : Condition, event : Event, captures : Array[Capture]) -> Bool raise CepError

    Validate untrusted mutable condition arrays before standalone evaluation. Recognition uses a private fast path over the Pattern's sealed owned AST.

    Condition::explain

    source
    fn Condition::explain(self : Condition, event : Event, captures? : Array[Capture]) -> Evaluation raise CepError

    Return value-free diagnostic reasons. Field names and raw values are not copied into the notes; paths refer to condition AST positions.

    Condition::not_equal

    source
    fn Condition::not_equal(x : Condition, y : Condition) -> Bool

    Condition::to_repr

    source

    Condition::validate

    source
    fn Condition::validate(self : Condition, allowed_captures : Array[String]) -> Unit raise CepError

    Validates the complete AST, including branches that would short circuit. Depth is at most 32 and the unfolded node count is at most 512. These bounds also reject cyclic arrays before any copying or runtime evaluation occurs.

    ConditionNote

    source
    pub(all) struct ConditionNote {
      path : String
      passed : Bool
      reason : String
    }

    Contiguity

    source
    pub(all) enum Contiguity {
      Strict
      SkipNext
      SkipAny
    } derive(Eq, 
    Debug    )

    How events not accepted by the next positive occurrence are treated.

    Contiguity::equal

    source
    fn Contiguity::equal(Contiguity, Contiguity) -> Bool

    Contiguity::not_equal

    source
    fn Contiguity::not_equal(x : Contiguity, y : Contiguity) -> Bool

    Contiguity::to_repr

    source

    DetectionMetrics

    source
    pub struct DetectionMetrics {
      series : Map[String, DetectionSeries]
      max_series : Int
      by_key : Bool
      batches : Int64
      processed : Int64
      work : Int64
    }

    Bounded result telemetry. Per-key dimensions are opt-in because their cardinality usually exceeds the number of configured patterns.

    DetectionMetrics::batches

    source
    fn DetectionMetrics::batches(self : DetectionMetrics) -> Int64

    DetectionMetrics::fork

    source

    DetectionMetrics::new

    source
    fn DetectionMetrics::new(max_series? : Int, by_key? : Bool) -> DetectionMetrics raise CepError

    DetectionMetrics::observe

    source
    fn DetectionMetrics::observe(self : DetectionMetrics, batch : Batch) -> Unit raise CepError

    Observe a committed batch atomically. Re-observing a batch counts it again; delivery deduplication belongs to the caller, using Match::identity.

    DetectionMetrics::processed

    source
    fn DetectionMetrics::processed(self : DetectionMetrics) -> Int64

    DetectionMetrics::reset

    source
    fn DetectionMetrics::reset(self : DetectionMetrics) -> Unit

    DetectionMetrics::series

    source

    DetectionMetrics::work

    source
    fn DetectionMetrics::work(self : DetectionMetrics) -> Int64

    DetectionSeries

    source
    pub(all) struct DetectionSeries {
      pattern : String
      key : String?
      matches : Int64
      absence_matches : Int64
      timeouts : Int64
      shortest_duration : Int64?
      longest_duration : Int64?
      mean_duration : Double?
    }

    Engine

    source
    pub struct Engine {
      patterns : Array[Pattern]
      limits : Limits
      buffer : Array[Stamped]
      active : Array[Branch]
      watermark : Int64
      ordinal : Int64
    }

    Engine::active_states

    source
    fn Engine::active_states(self : Engine) -> Int

    Engine::advance_watermark

    source
    fn Engine::advance_watermark(self : Engine, requested : Int64) -> Batch raise CepError

    Close the event-time prefix through W. Any failure leaves the engine unchanged.

    Engine::buffered

    source
    fn Engine::buffered(self : Engine) -> Int

    Engine::checkpoint

    source
    fn Engine::checkpoint(self : Engine) -> Checkpoint

    Takes a detached checkpoint; later engine mutations do not change it.

    Engine::finish

    source
    fn Engine::finish(self : Engine) -> Batch raise CepError

    Close every possible event and deadline in the documented timestamp domain. A failed finish has the same rollback semantics as advance_watermark.

    Engine::fork

    source
    fn Engine::fork(self : Engine) -> Engine

    Creates an independent engine at the same committed point, including buffered observations, capture evidence and mutable condition child arrays.

    Engine::frontier

    source
    fn Engine::frontier(self : Engine, key? : String, limit? : Int) -> Frontier raise CepError

    Return a bounded diagnostic sample. Truncation never changes recognition.

    Engine::inspect

    source
    fn Engine::inspect(self : Engine) -> Inspection

    Snapshot current committed occupancy without including business field values.

    Engine::is_finished

    source
    fn Engine::is_finished(self : Engine) -> Bool

    Engine::limits

    source
    fn Engine::limits(self : Engine) -> Limits

    Engine::new

    source
    fn Engine::new(patterns : Array[Pattern], limits? : Limits) -> Engine raise CepError

    Engine::next_deadline

    source
    fn Engine::next_deadline(self : Engine) -> Int64?

    Earliest active deadline. The host still determines when event time is complete; this value must not be treated as a wall-clock watermark.

    Engine::next_event_time

    source
    fn Engine::next_event_time(self : Engine) -> Int64?

    Earliest buffered event, for host-driven watermark and batching policies.

    Engine::preview

    source
    fn Engine::preview(self : Engine, events : Array[Event], watermark : Int64, frontier_limit? : Int) -> Preview raise CepError

    Preview a batch of observations and one watermark against a detached engine. Existing buffered observations through that watermark are also processed. Neither success nor failure changes the original engine or its ordinals.

    Engine::push

    source
    fn Engine::push(self : Engine, event : Event) -> Int64 raise CepError

    Engine::push_many

    source
    fn Engine::push_many(self : Engine, events : Array[Event]) -> Array[Int64] raise CepError

    Accepts the entire batch or leaves the engine unchanged. In particular, a late event or an exhausted buffer cannot partially consume ordinals.

    Engine::watermark

    source
    fn Engine::watermark(self : Engine) -> Int64

    Engine::with_limits

    source
    fn Engine::with_limits(self : Engine, limits : Limits) -> Engine raise CepError

    Returns a detached engine with replacement budgets. Existing buffered events, branches and active keys must fit; output/work limits govern future calls. This supports retrying a failed advance with larger budgets.

    Evaluation

    source
    pub struct Evaluation {
      result : Bool
      notes : Array[ConditionNote]
    }

    Evaluation::matched

    source
    fn Evaluation::matched(self : Evaluation) -> Bool

    Evaluation::notes

    source

    Event

    source
    pub struct Event {
      id : String
      key : String
      time : Int64
      kind : String
      fields : Array[Field]
    } derive(Eq, 
    Debug    )

    Event::equal

    source
    fn Event::equal(Event, Event) -> Bool

    Event::field

    source
    fn Event::field(self : Event, name : String) -> Value?

    Event::fields

    source
    fn Event::fields(self : Event) -> Array[Field]

    Event::id

    source
    fn Event::id(self : Event) -> String

    Event::key

    source
    fn Event::key(self : Event) -> String

    Event::kind

    source
    fn Event::kind(self : Event) -> String

    Event::new

    source
    fn Event::new(id : String, key : String, time : Int64, kind : String, fields? : Array[Field]) -> Event raise CepError

    Event::not_equal

    source
    fn Event::not_equal(x : Event, y : Event) -> Bool

    Event::time

    source
    fn Event::time(self : Event) -> Int64

    Event::to_repr

    source

    Field

    source
    pub(all) struct Field {
      name : String
      value : Value
    } derive(Eq, 
    Debug    )

    Field::equal

    source
    fn Field::equal(Field, Field) -> Bool

    Field::not_equal

    source
    fn Field::not_equal(x : Field, y : Field) -> Bool

    Field::to_repr

    source

    Frontier

    source
    pub struct Frontier {
      items : Array[Pending]
      total : Int
    }

    Frontier::items

    source
    fn Frontier::items(self : Frontier) -> Array[Pending]

    Frontier::total

    source
    fn Frontier::total(self : Frontier) -> Int

    Frontier::truncated

    source
    fn Frontier::truncated(self : Frontier) -> Bool

    Inspection

    source
    pub struct Inspection {
      watermark : Int64
      occurrence : Int64
      buffered : Int
      active : Int
      partitions : Array[PartitionStatus]
      patterns : Array[PatternStatus]
    }

    Inspection::active_states

    source
    fn Inspection::active_states(self : Inspection) -> Int

    Inspection::buffered

    source
    fn Inspection::buffered(self : Inspection) -> Int

    Inspection::occurrence

    source
    fn Inspection::occurrence(self : Inspection) -> Int64

    Inspection::partitions

    source
    fn Inspection::partitions(self : Inspection) -> Array[PartitionStatus]

    Inspection::patterns

    source
    fn Inspection::patterns(self : Inspection) -> Array[PatternStatus]

    Inspection::watermark

    source
    fn Inspection::watermark(self : Inspection) -> Int64

    Limits

    source
    pub(all) struct Limits {
      buffered_events : Int
      active_states : Int
      partitions : Int
      outputs_per_call : Int
      work_per_call : Int
    } derive(Eq, 
    Debug    )

    Limits::default

    source
    fn Limits::default() -> Limits

    Limits::equal

    source
    fn Limits::equal(Limits, Limits) -> Bool

    Limits::not_equal

    source
    fn Limits::not_equal(x : Limits, y : Limits) -> Bool

    Limits::to_repr

    source

    Limits::validate

    source
    fn Limits::validate(self : Limits) -> Unit raise CepError

    Match

    source
    pub struct Match {
      pattern : String
      key : String
      captures : Array[Capture]
      occurrences : Array[Int64]
      started : Int64
      ended : Int64
      absence : Bool
    }

    Match::captures

    source
    fn Match::captures(self : Match) -> Array[Capture]

    Match::ended

    source
    fn Match::ended(self : Match) -> Int64

    Match::events

    source
    fn Match::events(self : Match, name : String) -> Array[Event]

    Match::identity

    source
    fn Match::identity(self : Match) -> String

    Identity is scoped to the same session and successful input occurrence order. Length prefixes prevent collisions caused by delimiters in user identifiers.

    Match::is_absence

    source
    fn Match::is_absence(self : Match) -> Bool

    Match::key

    source
    fn Match::key(self : Match) -> String

    Match::last_event_time

    source
    fn Match::last_event_time(self : Match) -> Int64

    Match::occurrences

    source
    fn Match::occurrences(self : Match) -> Array[Int64]

    Match::pattern

    source
    fn Match::pattern(self : Match) -> String

    Match::started

    source
    fn Match::started(self : Match) -> Int64

    Monitor

    source
    pub struct Monitor {
      reader : CommandReader
      session : Session
      accepted_events : Int64
      advances : Int64
      matches : Int64
      timeouts : Int64
      work : Int64
    }

    Owns the text-framing and detection transaction boundary for an application. Failed ingestion commits neither partial text nor journal or detector state.

    Monitor::finish_input

    source
    fn Monitor::finish_input(self : Monitor, close_time? : Bool) -> Array[Batch] raise CepError

    Finish the final non-newline-terminated command. By default also close all event-time deadlines. A repeated successful finish returns no new batches.

    Monitor::fork

    source
    fn Monitor::fork(self : Monitor) -> Monitor

    Monitor::ingest

    source
    fn Monitor::ingest(self : Monitor, chunk : String) -> Array[Batch] raise CepError

    Monitor::inspection

    source
    fn Monitor::inspection(self : Monitor) -> Inspection

    Monitor::new

    source
    fn Monitor::new(patterns : Array[Pattern], limits? : Limits, max_commands? : Int, max_line? : Int, commands_per_chunk? : Int) -> Monitor raise CepError

    Monitor::session

    source
    fn Monitor::session(self : Monitor) -> Session

    Monitor::stats

    source
    fn Monitor::stats(self : Monitor) -> MonitorStats

    MonitorStats

    source
    pub(all) struct MonitorStats {
      accepted_events : Int64
      advances : Int64
      matches : Int64
      timeouts : Int64
      work : Int64
      buffered_events : Int
      active_states : Int
      pending_text_units : Int
    }

    PartitionStatus

    source
    pub(all) struct PartitionStatus {
      key : String
      buffered : Int
      active : Int
      earliest_event : Int64?
      latest_event : Int64?
      earliest_deadline : Int64?
    }

    Pattern

    source
    pub struct Pattern {
      id : String
      stages : Array[Stage]
      within : Int64
      after_match : AfterMatch
      absence : Condition?
    }

    A bounded pattern with a mandatory first stage and unique capture names.

    Pattern::absence

    source
    fn Pattern::absence(self : Pattern) -> Condition?

    Pattern::after_match

    source
    fn Pattern::after_match(self : Pattern) -> AfterMatch

    Pattern::analysis

    source
    fn Pattern::analysis(self : Pattern) -> PatternAnalysis

    Pattern::diagnose

    source
    fn Pattern::diagnose(self : Pattern, event : Event, captures? : Array[Capture]) -> Array[StageDiagnostic] raise CepError

    Explain each stage's predicates against supplied capture context. This is a predicate diagnostic, not a claim that the event can complete the sequence.

    Pattern::id

    source
    fn Pattern::id(self : Pattern) -> String

    Pattern::new

    source
    fn Pattern::new(id : String, stages : Array[Stage], within : Int64, after_match? : AfterMatch, absence? : Condition) -> Pattern raise CepError

    Pattern::stage

    source
    fn Pattern::stage(self : Pattern, name : String) -> Stage?

    Pattern::stage_at

    source
    fn Pattern::stage_at(self : Pattern, index : Int) -> Stage?

    Pattern::stage_names

    source
    fn Pattern::stage_names(self : Pattern) -> Array[String]

    Pattern::stages

    source
    fn Pattern::stages(self : Pattern) -> Array[Stage]

    Pattern::within

    source
    fn Pattern::within(self : Pattern) -> Int64

    PatternAnalysis

    source
    pub(all) struct PatternAnalysis {
      minimum_captures : Int
      maximum_captures : Int
      nullable_stages : Int
      variable_stages : Int
      repeated_stages : Int
      strict_stages : Int
      skip_any_stages : Int
      guarded_stages : Int
      has_absence : Bool
      branching_risk : BranchingRisk
    } derive(Eq, 
    Debug    )

    PatternAnalysis::equal

    source

    PatternAnalysis::not_equal

    source
    fn PatternAnalysis::not_equal(x : PatternAnalysis, y : PatternAnalysis) -> Bool

    PatternAnalysis::to_repr

    source

    PatternStatus

    source
    pub(all) struct PatternStatus {
      pattern : String
      active : Int
      awaiting_absence : Int
      earliest_deadline : Int64?
    }

    Pending

    source
    pub(all) struct Pending {
      pattern : String
      key : String
      stage : String?
      matched_here : Int
      minimum_more : Int
      maximum_more : Int
      deadline : Int64
      awaiting_absence : Bool
      captures : Array[Capture]
    }

    Preview

    source
    pub struct Preview {
      batch : Batch
      inspection : Inspection
      frontier : Frontier
    }

    Preview::batch

    source
    fn Preview::batch(self : Preview) -> Batch

    Preview::frontier

    source
    fn Preview::frontier(self : Preview) -> Frontier

    Preview::inspection

    source
    fn Preview::inspection(self : Preview) -> Inspection

    Program

    source
    pub struct Program {
      patterns : Array[Pattern]
      limits : Limits
      projections : Array[ProjectionPlan]
    }

    A validated deployable set of patterns, budgets and optional alert projections.

    Program::describe

    source
    fn Program::describe(self : Program) -> Json

    Program::engine

    source
    fn Program::engine(self : Program) -> Engine raise CepError

    Program::limits

    source
    fn Program::limits(self : Program) -> Limits

    Program::monitor

    source
    fn Program::monitor(self : Program, max_commands? : Int) -> Monitor raise CepError

    Program::new

    source
    fn Program::new(patterns : Array[Pattern], limits? : Limits, projections? : Array[ProjectionPlan]) -> Program raise CepError

    Program::patterns

    source
    fn Program::patterns(self : Program) -> Array[Pattern]

    Program::project

    source
    fn Program::project(self : Program, batch : Batch) -> Array[AlertRecord] raise CepError

    Project only matches with a configured projection. Failure returns no partial records and never changes the already committed recognition batch.

    Program::projections

    source
    fn Program::projections(self : Program) -> Array[ProjectionPlan]

    Projection

    source
    pub(all) enum Projection {
      EntityKey
      PatternName
      StartTime
      EndTime
      Duration
      AbsenceMatch
      CaptureSize(String)
      FirstField(String, String)
      LastField(String, String)
      TotalInteger(String, String)
      TotalDecimal(String, String)
      MinimumField(String, String)
      MaximumField(String, String)
      AverageDecimal(String, String)
    } derive(Eq, 
    Debug    )

    Projection::equal

    source
    fn Projection::equal(Projection, Projection) -> Bool

    Projection::not_equal

    source
    fn Projection::not_equal(x : Projection, y : Projection) -> Bool

    Projection::to_repr

    source

    ProjectionPlan

    source
    pub struct ProjectionPlan {
      pattern : String
      columns : Array[Column]
    }

    ProjectionPlan::apply

    source
    fn ProjectionPlan::apply(self : ProjectionPlan, value : Match) -> Array[Field] raise CepError

    ProjectionPlan::columns

    source
    fn ProjectionPlan::columns(self : ProjectionPlan) -> Array[Column]

    ProjectionPlan::new

    source
    fn ProjectionPlan::new(pattern : Pattern, columns : Array[Column]) -> ProjectionPlan raise CepError

    Session

    source
    pub struct Session {
      engine : Engine
      commands : Array[StreamInput]
      max_commands : Int
    }

    A bounded input journal coupled atomically to an in-memory recognizer. Replaying a journal reconstructs detection state, not external sink delivery.

    Session::active_states

    source
    fn Session::active_states(self : Session) -> Int

    Session::apply

    source
    fn Session::apply(self : Session, command : StreamInput) -> Update raise CepError

    Session::apply_many

    source
    fn Session::apply_many(self : Session, commands : Array[StreamInput]) -> Array[Update] raise CepError

    Session::buffered

    source
    fn Session::buffered(self : Session) -> Int

    Session::checkpoint

    source
    fn Session::checkpoint(self : Session) -> Checkpoint

    Session::commands

    source
    fn Session::commands(self : Session) -> Array[StreamInput]

    Session::finish

    source
    fn Session::finish(self : Session) -> Batch raise CepError

    Session::fork

    source
    fn Session::fork(self : Session) -> Session

    Session::new

    source
    fn Session::new(patterns : Array[Pattern], limits? : Limits, max_commands? : Int) -> Session raise CepError

    Session::watermark

    source
    fn Session::watermark(self : Session) -> Int64

    Session::with_limits

    source
    fn Session::with_limits(self : Session, limits : Limits) -> Session raise CepError

    Stage

    source
    pub struct Stage {
      name : String
      condition : Condition
      min : Int
      max : Int
      contiguity : Contiguity
      forbidden : Condition?
    }

    One named positive requirement. Construction detaches condition arrays.

    Stage::condition

    source
    fn Stage::condition(self : Stage) -> Condition

    Stage::contiguity

    source
    fn Stage::contiguity(self : Stage) -> Contiguity

    Stage::forbidden

    source
    fn Stage::forbidden(self : Stage) -> Condition?

    Stage::is_optional

    source
    fn Stage::is_optional(self : Stage) -> Bool

    Stage::is_repeated

    source
    fn Stage::is_repeated(self : Stage) -> Bool

    Stage::maximum

    source
    fn Stage::maximum(self : Stage) -> Int

    Stage::minimum

    source
    fn Stage::minimum(self : Stage) -> Int

    Stage::name

    source
    fn Stage::name(self : Stage) -> String

    Stage::new

    source
    fn Stage::new(name : String, condition : Condition, min? : Int, max? : Int, contiguity? : Contiguity, forbidden? : Condition) -> Stage raise CepError

    StageDiagnostic

    source
    pub(all) struct StageDiagnostic {
      stage : String
      positive : Evaluation
      forbidden : Evaluation?
      minimum : Int
      maximum : Int
    }

    Stamped

    source
    type Stamped

    StreamInput

    source
    pub(all) enum StreamInput {
      Observe(Event)
      Advance(Int64)
    } derive(Eq, 
    Debug    )

    A command for deterministic, explicitly watermarked in-memory replay.

    StreamInput::equal

    source
    fn StreamInput::equal(StreamInput, StreamInput) -> Bool

    StreamInput::not_equal

    source
    fn StreamInput::not_equal(x : StreamInput, y : StreamInput) -> Bool

    StreamInput::to_repr

    source

    Timeout

    source
    pub struct Timeout {
      pattern : String
      key : String
      deadline : Int64
      captures : Array[Capture]
    }

    Timeout::captures

    source
    fn Timeout::captures(self : Timeout) -> Array[Capture]

    Timeout::deadline

    source
    fn Timeout::deadline(self : Timeout) -> Int64

    Timeout::key

    source
    fn Timeout::key(self : Timeout) -> String

    Timeout::pattern

    source
    fn Timeout::pattern(self : Timeout) -> String

    Update

    source
    pub(all) enum Update {
      Accepted(Int64)
      Progress(Batch)
    }

    Value

    source
    pub(all) enum Value {
      Text(String)
      Integer(Int64)
      Decimal(Double)
      Boolean(Bool)
      NullValue
    } derive(Eq, 
    Debug    )

    Value::compare

    source
    fn Value::compare(self : Value, comparison : Comparison, other : Value) -> Bool

    Comparisons never coerce integer values to floating point.

    Value::equal

    source
    fn Value::equal(Value, Value) -> Bool

    Value::not_equal

    source
    fn Value::not_equal(x : Value, y : Value) -> Bool

    Value::to_repr

    source

    Value::type_name

    source
    fn Value::type_name(self : Value) -> String

    Value::validate

    source
    fn Value::validate(self : Value) -> Unit raise CepError

    FINAL_WATERMARK

    source
    let FINAL_WATERMARK : Int64

    MAX_CAPTURED_EVENTS

    source
    let MAX_CAPTURED_EVENTS : Int

    MAX_CONDITION_DEPTH

    source
    let MAX_CONDITION_DEPTH : Int

    MAX_CONDITION_NODES

    source
    let MAX_CONDITION_NODES : Int

    MAX_EVENT_FIELDS

    source
    let MAX_EVENT_FIELDS : Int

    MAX_EVENT_TIME

    source
    let MAX_EVENT_TIME : Int64

    MAX_PATTERNS

    source
    let MAX_PATTERNS : Int

    MAX_STAGES

    source
    let MAX_STAGES : Int

    MAX_STAGE_REPETITIONS

    source
    let MAX_STAGE_REPETITIONS : Int

    MAX_VALUE_TEXT_UNITS

    source
    let MAX_VALUE_TEXT_UNITS : Int

    MAX_WINDOW

    source
    let MAX_WINDOW : Int64

    MAX_WIRE_TEXT_UNITS

    source
    let MAX_WIRE_TEXT_UNITS : Int

    alert_record_to_json

    source
    fn alert_record_to_json(record : AlertRecord) -> Json raise CepError

    batch_to_json

    source
    fn batch_to_json(batch : Batch) -> Json

    checked_format_event

    source
    fn checked_format_event(event : Event) -> String raise CepError

    Format an event only when its escaped JSON fits the bounded text parser.

    command_chunks

    source
    fn command_chunks(text : String, max_units? : Int) -> Iter[StringView] raise CepError

    Chunk a complete Unicode string without discarding a scalar at a UTF-16 boundary. A two-unit scalar can exceed a requested size of one by one unit.

    command_from_json

    source
    fn command_from_json(value : Json) -> StreamInput raise CepError

    command_to_json

    source
    fn command_to_json(command : StreamInput) -> Json

    condition_from_json

    source
    fn condition_from_json(value : Json) -> Condition raise CepError

    condition_to_json

    source
    fn condition_to_json(condition : Condition) -> Json raise CepError

    evaluation_to_json

    source
    fn evaluation_to_json(evaluation : Evaluation) -> Json

    event_from_json

    source
    fn event_from_json(value : Json) -> Event raise CepError

    event_to_json

    source
    fn event_to_json(event : Event) -> Json

    Event timestamps and integer field values are encoded as decimal strings.

    format_command

    source
    fn format_command(command : StreamInput) -> String raise CepError

    format_condition

    source
    fn format_condition(condition : Condition) -> String raise CepError

    format_event

    source
    fn format_event(event : Event) -> String

    Format any valid event. The text parser separately enforces a 1 Mi-unit input limit.

    format_pattern

    source
    fn format_pattern(pattern : Pattern) -> String raise CepError

    format_program

    source
    fn format_program(program : Program) -> String raise CepError

    format_session

    source
    fn format_session(session : Session) -> String raise CepError

    inspection_to_json

    source
    fn inspection_to_json(value : Inspection) -> Json

    limits_from_json

    source
    fn limits_from_json(value : Json) -> Limits raise CepError

    limits_to_json

    source
    fn limits_to_json(limits : Limits) -> Json

    match_to_json

    source
    fn match_to_json(value : Match) -> Json

    metrics_to_json

    source
    fn metrics_to_json(metrics : DetectionMetrics) -> Json

    monitor_stats_to_json

    source
    fn monitor_stats_to_json(stats : MonitorStats) -> Json

    parse_command

    source
    fn parse_command(text : String) -> StreamInput raise CepError

    parse_condition

    source
    fn parse_condition(text : String) -> Condition raise CepError

    Wire nesting allows object/array wrappers around the 32-level AST.

    parse_event

    source
    fn parse_event(text : String) -> Event raise CepError

    parse_pattern

    source
    fn parse_pattern(text : String) -> Pattern raise CepError

    parse_program

    source
    fn parse_program(text : String) -> Program raise CepError

    parse_session

    source
    fn parse_session(text : String, max_work? : Int) -> Session raise CepError

    pattern_from_json

    source
    fn pattern_from_json(value : Json) -> Pattern raise CepError

    pattern_to_json

    source
    fn pattern_to_json(pattern : Pattern) -> Json

    program_from_json

    source
    fn program_from_json(value : Json) -> Program raise CepError

    program_to_json

    source
    fn program_to_json(program : Program) -> Json

    projected_record_to_json

    source
    fn projected_record_to_json(fields : Array[Field]) -> Json raise CepError

    projection_from_json

    source
    fn projection_from_json(pattern : Pattern, value : Json) -> ProjectionPlan raise CepError

    projection_to_json

    source
    fn projection_to_json(plan : ProjectionPlan) -> Json

    replay

    source
    fn replay(patterns : Array[Pattern], commands : Array[StreamInput], limits? : Limits, max_commands? : Int, max_outputs? : Int) -> Array[Batch] raise CepError

    Replays exactly the supplied commands and returns one batch per Advance. There is no implicit final watermark. Each Advance has its own engine work budget; replay separately bounds command count and aggregate result count. On failure no intermediate batches escape, and input objects are unchanged.

    session_from_json

    source
    fn session_from_json(value : Json, max_work? : Int) -> Session raise CepError

    Restore by validating and replaying committed inputs. max_work is a total import budget, in addition to each engine advance's own work limit.

    session_to_json

    source
    fn session_to_json(session : Session) -> Json

    stage_diagnostics_to_json

    source
    fn stage_diagnostics_to_json(diagnostics : Array[StageDiagnostic]) -> Json

    timeout_to_json

    source
    fn timeout_to_json(value : Timeout) -> Json

    value_from_json

    source
    fn value_from_json(value : Json) -> Value raise CepError

    value_to_json

    source
    fn value_to_json(value : Value) -> Json raise CepError

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io