Query and follow event streams

Flowcore stores events in time buckets and identifies positions within a bucket by event ID. Querying and streaming build on those two coordinates.

Use direct queries when an application needs a bounded set of retained events. Use the CLI for investigation and ad hoc streaming. Use a Data Pump or Data Pathway for durable application processing.

Understand time buckets

A time bucket is a time-based storage partition. It keeps event history queryable without treating the complete lifetime of an event type as one unbounded response.

A durable position consists of:

• A timeBucket
• An eventId within that bucket

Consumers move through events in the current bucket, then advance to the next available bucket. Persisting both values allows processing to resume after restarts.

Do not treat a timestamp alone as a precise checkpoint. Multiple events can share a time range, and late or concurrent writes can make timestamp-only recovery ambiguous.

Choose ID-based or name-based queries

The SDK supports both resource-ID and resource-name query patterns.

Use IDs when your application already resolved resources and wants an unambiguous platform identifier. Use names when configuration is naturally expressed as tenant, data core, flow type, and event type names.

Avoid resolving names separately for every page. Cache stable resource identifiers appropriately or use a name-based command designed for repeated fetching.

List available time buckets

TimeBucketListCommand lists buckets for one or more event type IDs:

import {
  FlowcoreClient,
  TimeBucketListCommand,
} from "@flowcore/sdk"
 
const client = new FlowcoreClient({
  apiKey: process.env.FLOWCORE_API_KEY!,
})
 
const result = await client.execute(new TimeBucketListCommand({
  tenant: "acme-production",
  eventTypeId: "et_replace_me",
  fromTimeBucket: "20260701h000000",
  toTimeBucket: "20260731h235959",
  order: "asc",
  pageSize: 100,
}))
 
console.log(result.timeBuckets)
console.log(result.nextCursor)

Use ascending order for forward processing and replay. If nextCursor is returned, request the next page until no cursor remains.

Time-bucket cursors and event-page cursors are different types of progress. Keep them separate in code and persisted state.

List buckets by resource names

EventsFetchTimeBucketsByNamesCommand resolves available buckets from names:

import { EventsFetchTimeBucketsByNamesCommand } from "@flowcore/sdk"
 
const result = await client.execute(
  new EventsFetchTimeBucketsByNamesCommand({
    tenant: "acme-production",
    dataCoreId: "dc_replace_me",
    flowType: "order.0",
    eventTypes: ["order.placed.0", "order.cancelled.0"],
    fromTimeBucket: "20260701h000000",
    pageSize: 100,
  }),
)

This is useful when a consumer follows several related event types from the same flow type.

Fetch events from a bucket

EventListCommand fetches events for one or more event type IDs:

import { EventListCommand } from "@flowcore/sdk"
 
let cursor: string | undefined
 
do {
  const page = await client.execute(new EventListCommand({
    tenant: "acme-production",
    eventTypeId: ["et_order_placed", "et_order_cancelled"],
    timeBucket: "20260715h120000",
    pageSize: 500,
    cursor,
    order: "asc",
    includeSensitiveData: false,
  }))
 
  for (const event of page.events) {
    await processEvent(event)
  }
 
  cursor = page.nextCursor
} while (cursor)

Process a bounded page and commit progress before requesting too far ahead. Very large prefetch buffers increase memory use and the amount of work repeated after failure.

includeSensitiveData should remain false unless the caller is authorized and the use case requires protected values. Do not request sensitive data merely to simplify one handler.

Fetch events by names

EventsFetchCommand fetches events using the data core ID plus flow and event type names:

import { EventsFetchCommand } from "@flowcore/sdk"
 
const page = await client.execute(new EventsFetchCommand({
  tenant: "acme-production",
  dataCoreId: "dc_replace_me",
  flowType: "order.0",
  eventTypes: ["order.placed.0", "order.cancelled.0"],
  timeBucket: "20260715h120000",
  pageSize: 500,
  afterEventId: lastProcessedEventId,
  includeSensitiveData: false,
}))

Use afterEventId to resume after a fully processed event. fromEventId and afterEventId have different inclusion semantics and must not be supplied together.

Use toEventId when a bounded operation must stop at a known event.

Preserve ordering assumptions

Ascending reads are the normal choice for projections and replay. Descending reads are useful for inspection, such as showing recent events, but the SDK notes that descending mode disables pagination and event-range filters for EventListCommand.

Do not build a durable consumer around descending pages.

When processing more than one event type, define whether ordering across those event types matters. Concurrency can improve throughput but may violate assumptions about sequential state transitions.

If a handler requires strict per-entity ordering, encode and enforce that requirement explicitly rather than assuming global delivery order.

Persist progress safely

A consumer should advance its stored position only after required work is durable.

For a database projection:

1. Begin a transaction.
2. Check or insert the event ID as a deduplication record.
3. Apply the projection change.
4. Commit.
5. Persist the safe bucket and event position according to the pump or consumer design.

If progress advances before the projection commits, a crash can skip work. If the projection commits before progress advances, the event may be delivered again; idempotency handles that case.

At-least-once processing prefers safe duplicate delivery over silent loss.

Inspect streams with the CLI

The Flowcore CLI can stream historical and live events to the terminal:

flowcore stream \
  "https://flowcore.io/acme-production/commerce/order.0/order.placed.0.stream" \
  --start 1h \
  --json

Common start values include a relative duration, now, first, or an ISO timestamp. Use a maximum-event limit when exploring a high-volume stream.

The CLI also supports flow or event wildcards and lists of event types. Resolve broad patterns carefully in production tenants; a wildcard can produce far more data than expected.

To send events to an HTTP endpoint during controlled testing:

flowcore stream http \
  "https://flowcore.io/acme-development/commerce/order.0/order.placed.0.stream" \
  --start 10m \
  --destination http://localhost:3000/flowcore/events

Add authentication headers when the destination requires them. Do not stream sensitive production history to an untrusted local endpoint.

CLI streaming is valuable for diagnosis and experiments. A production consumer should use a managed or virtual Data Pathway with durable state and operational monitoring.

Use notifications as wake-up signals

A Pathways Data Pump can discover new events through:

• WebSocket notifications, the default low-latency option
• NATS notifications when that infrastructure is available
• Polling at a configured interval

A notification tells the pump that work may be available. The durable source remains Flowcore event storage, and the persisted bucket/event position determines what to fetch.

This distinction is important: losing a transient WebSocket notification should not mean losing the event. After reconnecting, the pump continues from its durable position and queries retained history.

Do not store business state only in a live connection callback.

Select a notifier for the environment

Use WebSocket notifications when outbound persistent connections are supported and low latency matters.

Use NATS when the deployment already operates NATS and wants notification delivery through that infrastructure.

Use polling when persistent notification connections are unavailable or simplicity matters more than immediate wake-up. Choose an interval that balances latency and API load.

Regardless of notifier, monitor reconnects, polling failures, checkpoint age, and delivery lag.

Distinguish observation from durable processing

Three common tools serve different purposes:

Bounded SDK query

Use for audit views, support tools, migration utilities, and explicitly bounded reads.

CLI stream

Use for developer inspection, production diagnosis with appropriate authorization, and controlled HTTP testing.

Data Pump or managed Data Pathway

Use for continuous application processing, persisted progress, retries, backpressure, and replay.

Do not turn a support-page query into an unbounded background consumer. Do not treat a terminal stream as a production integration.

Diagnose missing or repeated events

If a query returns no events:

• Verify the tenant, data core, flow type, and event type.
• Confirm the requested time-bucket range.
• Check whether the caller may read the resource.
• Verify that sensitive-data options are allowed.
• Confirm the producer successfully ingested events.

If events repeat:

• Check whether the stored checkpoint advanced.
• Verify that the event ID is persisted after successful work.
• Look for a restart or reset to an earlier position.
• Confirm multiple consumers are not sharing state incorrectly.
• Make the handler idempotent rather than assuming retries will disappear.

If live updates stop:

• Check WebSocket, NATS, or poller health.
• Compare the checkpoint with the newest available bucket.
• Confirm the process is still fetching retained events after reconnect.
• Inspect permission and credential changes.

Continue with Replay history and rebuild projections for controlled historical reprocessing.