Querying

Events remain searchable after ACK. Query by payload fields, lineage metadata, and sequence bounds — without touching consumer-group state.

Filters and bounds

{:ok, eur} =
  Rheo.query("market-events",
    type: "curve_update",
    currency: "EUR",
    limit: 50
  )

{:ok, mid} =
  Rheo.query("market-events",
    after_sequence: 10,
    until_sequence: 15,
    order_by: [sequence: :asc]
  )

{:ok, by_corr} =
  Rheo.query("market-events", correlation_id: "corr-1", limit: 20)

Filters are portable across backends via Rheo.Query (see ADR 013). Capability flags describe what each backend indexes well; fencing semantics stay the same.

Pagination

query_page/2 returns a composite cursor (%{partition => after_sequence}) so a page never splits a sequence across partitions:

{:ok, page1} = Rheo.query_page("market-events", type: "curve_update", limit: 100)
# page1.next_cursor => %{0 => 12, 1 => 11, ...}

{:ok, page2} =
  Rheo.query_page("market-events",
    type: "curve_update",
    limit: 100,
    cursor: page1.next_cursor
  )

Streaming

"market-events"
|> Rheo.stream_query(type: "curve_update", limit: 200)
|> Stream.map(& &1.id)
|> Enum.take(1_000)

stream_query/2 pages internally with a bounded page size.

Lineage helpers

alias Rheo.Event.Lineage

Lineage.get(event, :correlation_id)
Lineage.get(event, :producer)

Read vs query vs fetch

Deeper walkthrough: Searching the stream.