Streaming Queries

Streaming lets you iterate over large component datasets record by record without loading everything into memory at once. The server emits results as NDJSON (newline-delimited JSON) — a metadata line first, followed by one record per line.

---

When to use streaming

Use case	Recommendation
Paginated UI — fetching a page of results	find() (standard)
Processing all records — exports, migrations, background jobs	find(..., { stream: true })
Large datasets where memory is a concern	find(..., { stream: true })

---

Usage

Pass { stream: true } as the second argument to find(). Instead of returning a Promise, it returns a FindChainable — a fluent builder you configure before calling .stream() to begin the request.

await platform.component('orders')
    .find({ currentPage: 0, perPage: 100 }, { stream: true })
    .onMeta((meta) => {
        console.log(`Total records: ${meta.total}`)
        console.log(`Total pages:   ${meta.total_pages}`)
    })
    .onData((record) => {
        // Called once per record
        console.log('Record:', record)
    })
    .onError((err) => {
        console.error('Stream error:', err)
    })
    .onEnd(() => {
        console.log('Stream complete')
    })
    .stream()

Note: .stream() starts the HTTP request and returns a Promise<void> that resolves when the stream ends. Always await it (or chain .catch()) to handle errors.

---

Chainable methods

Method	Callback signature	Description
.onMeta(cb)	(meta: FindStreamMeta) => void \| Promise<void>	Called once with pagination metadata before any records
.onData(cb)	(record: Model) => void \| Promise<void>	Called once per record
.onError(cb)	(error: Error) => void	Called if the stream fails
.onEnd(cb)	() => void	Called when all records have been emitted
.stream()	—	Starts the request; returns Promise<void>

All methods return this, so they can be chained in any order before calling .stream().

---

FindStreamMeta

The first line emitted by the server contains metadata:

type FindStreamMeta = {
    total: number           // total records matching the filter (before pagination)
    total_filtered: number  // same as total when a filter is applied
    total_pages: number     // total number of pages at the requested perPage
    page: number            // current page index (0-based)
    next_cursor?: string    // opaque cursor token for the next page (absent on last page)
}

---

Working with async callbacks

Both onMeta and onData callbacks support async functions. The stream processor awaits each callback before processing the next line, so back-pressure is naturally applied — you won't receive the next record until your handler resolves.

await platform.component('orders')
    .find({ perPage: 500 }, { stream: true })
    .onData(async (record) => {
        await processRecord(record) // stream pauses until this resolves
    })
    .stream()

---

Pagination within a stream

perPage controls how many records the server fetches per internal cursor page. For most use cases the default of 10 is too small — set a larger value to reduce round-trips:

await platform.component('orders')
    .find({ perPage: 500, filter: 'status:paid' }, { stream: true })
    .onData((record) => { /* ... */ })
    .stream()

---

TypeScript types

The following types are exported from @ptkl/sdk:

import type { FindStreamMeta, FindStreamMetaCallback, FindChainable } from '@ptkl/sdk'

Type	Description
FindStreamMeta	Metadata object emitted before the first record
FindStreamMetaCallback	(meta: FindStreamMeta) => void \| Promise<void>
FindChainable	Fluent builder returned by find(..., { stream: true })

---

HTTP API

The streaming endpoint is available directly if you are not using the SDK:

POST /v4/system/component/{ref}/models/stream
Accept: application/x-ndjson
Content-Type: application/json

Request body — same shape as the standard find endpoint:

{
    "offset": 0,
    "limit": 100,
    "filter": "",
    "filterOn": [],
    "sortBy": "CreatedAt",
    "sortDesc": -1
}

Response — NDJSON stream. First line is metadata, subsequent lines are records, and an optional final line contains the cursor for the next page:

{"total":1500,"total_filtered":1500,"total_pages":15,"page":0}
{"uuid":"...","data":{...},"CreatedAt":"..."}
{"uuid":"...","data":{...},"CreatedAt":"..."}
...
{"next_cursor":"eyJ2IjoiMjAyNi0wNi0yMlQwOToxMjowMFoiLCJpZCI6ImFiYy0xMDAifQ"}

The next_cursor line is only emitted when there are more pages. It contains exactly one key (next_cursor) so it can be distinguished from record lines.

---

Cursor pagination with streaming

Cursor-based pagination works with the streaming endpoint the same way as with the standard Find API. Pass the cursor from a previous response to resume from where you left off:

// Page 1
let nextCursor = null

await platform.component('orders')
    .find({ perPage: 100, sortBy: 'UpdatedAt', sortDesc: -1 }, { stream: true })
    .onData((record) => { /* process record */ })
    .onCursor((cursor) => { nextCursor = cursor.next_cursor })
    .stream()

// Page 2 — pass the cursor
if (nextCursor) {
    await platform.component('orders')
        .find({ perPage: 100, sortBy: 'UpdatedAt', sortDesc: -1, cursor: nextCursor }, { stream: true })
        .onData((record) => { /* process record */ })
        .onCursor((cursor) => { nextCursor = cursor.next_cursor })
        .stream()
}

HTTP API example

Page 1:

POST /v4/system/component/{ref}/models/stream
Content-Type: application/json
Accept: application/x-ndjson

{ "offset": 0, "limit": 100, "sortBy": "UpdatedAt", "sortDesc": -1 }

Page 2 — pass the cursor from the previous response:

{ "offset": 0, "limit": 100, "sortBy": "UpdatedAt", "sortDesc": -1, "cursor": "eyJ2IjoiMjAyNi0wNi0yMlQwOToxMjowMFoiLCJpZCI6ImFiYy0xMDAifQ" }

When cursor is present the server uses an indexed range query to seek directly to the correct position instead of scanning and skipping records. See cursor pagination for details and important notes on sort/filter consistency.