Cross-Stream Relationships

Join, correlate, detect sequences, and traverse causal chains across streams.

enrich() — Left Join by Key

Join data from another stream using a matching key. For each record, Liven looks up the source_stream by the join_key value and merges the matched record's fields into the result.

from("orders") | enrich("customers", "customer_id")

This enriches each order with the customer's details from the customers stream. If no matching customer is found, the original record is unchanged.

correlate() — Time-Bounded Stream Join

Join records from two streams on a shared key within a time window. Useful for finding relationships between events that occur near each other in time.

from("transactions") | correlate("logins", "user_id", within: 60000)

This returns transactions that have a matching login event within ±60 seconds (±30s before to ±30s after). Matched records are enriched with correlated event data.

Fraud Detection Example

Find high-value transactions with no login in the preceding 5 minutes:

from("transactions") | filter(amount > 1000) | correlate("logins", "user_id", within: 300000)

sequence() — Event Pattern Detection

Detect ordered sequences of events within a time window using a finite state machine:

from("telemetry") | sequence(
  event == "cpu_spike",
  then: event == "memory_leak",
  within: 30000
)

The FSM steps through each condition in order. If all are satisfied within the time window, the final matching record is emitted. The window resets on expiry or completion.

Brute Force Detection

Detect three consecutive failed logins within 60 seconds:

from("auth") | sequence(
  action == "login_fail",
  then: action == "login_fail",
  then: action == "login_fail",
  within: 60000
)

Maximum 10 steps per sequence pattern. The sequence resets when the window expires.

chain() — Causal Chain Traversal

Follow relationships across multiple streams hop by hop. Each chain stage remaps the join key:

from("prompts") | chain("responses", "prompt_id")

This follows the relationship from prompts to responses by matching prompt_id in both streams. Chain is a left join — records without a match retain their original value.

Multi-Hop Chain

Traverse multiple levels deep by chaining stages:

from("prompts") | chain("responses", "prompt_id") | chain("memory", "response_id")

Each hop follows the relationship one level deeper. Useful for AI memory linking, transaction lineage, and multi-step event tracing.