View as Markdown

Reading & change feed

Two endpoints read a stream in cursor order: /read returns entries (what is in the stream), /changes returns inserts and erasures (what happened to it). Both share the cursor space, limits and pagination rules.

#/read

H="authorization: Bearer $CAPLESS_API_KEY"

curl -s https://api.capless.sh/v1/streams/events/read -H "$H" -d '{
  "afterCursor": null,
  "filter": "entry.amount > 50",
  "map": "{\"user\": entry.user_id, \"amount\": entry.amount}"
}'

{
  "entries": [{ "id": "ent_…", "cursor": "cur_…", "value": { "user": "u2", "amount": 200 } }],
  "readEntries": 2, "readBytes": 81, "resultEntries": 1, "resultBytes": 64, "errorEntries": 0,
  "nextCursor": "cur_…", "snapshotCursor": "cur_…", "hasMore": false
}

• Items are returned in ascending cursor order, strictly between afterCursor and beforeCursor (both exclusive boundaries).
• filter and map are CEL and run server-side. Mapped values keep the original id and cursor alongside value.
• Erased entries are never returned, even before an erasure is resolved.

#Paging without skips or repeats

1. Omit beforeCursor on the first page. The head at request start is captured and returned as snapshotCursor.
2. Pass nextCursor as afterCursor and snapshotCursor as beforeCursor on later pages for a stable snapshot.
3. Stop when hasMore is false. Then nextCursor == snapshotCursor.

AFTER=null; BEFORE=null
while :; do
  PAGE=$(curl -s https://api.capless.sh/v1/streams/events/read -H "$H" \
    -d "{\"afterCursor\":$AFTER,\"beforeCursor\":$BEFORE}")
  echo "$PAGE" | node -e 'for (const e of JSON.parse(require("fs").readFileSync(0)).entries) console.log(JSON.stringify(e.value))'
  [ "$(echo "$PAGE" | node -p 'JSON.parse(require("fs").readFileSync(0)).hasMore')" = true ] || break
  AFTER=$(echo "$PAGE" | node -p 'JSON.stringify(JSON.parse(require("fs").readFileSync(0)).nextCursor)')
  BEFORE=$(echo "$PAGE" | node -p 'JSON.stringify(JSON.parse(require("fs").readFileSync(0)).snapshotCursor)')
done

nextCursor is the boundary after the last examined position: everything before it has been examined. A page may return zero entries and still advance (for example, everything was filtered out), so always continue from nextCursor, never from the last returned entry. To tail a stream, keep calling with afterCursor = your last nextCursor and no beforeCursor.

#Limits

Limits end a page; they never drop data. A matching entry that does not fit in limitBytes is not consumed; the page ends before it. If even the first entry cannot fit, the request fails (422 result_too_large / limit_too_small) instead of skipping data.

#Evaluation errors

A CEL error on any entry (for example a missing field) fails the request with 422 expression_evaluation_failed, including the entry's ID and cursor. With "onError": "exclude", such entries are skipped and counted in errorEntries.

curl -s https://api.capless.sh/v1/streams/web.events/read -H "$H" \
  -d '{"filter":"entry.geo.country == \"DE\"","onError":"exclude"}'

Or guard with has(): has(entry.geo) && entry.geo.country == "DE".

#/changes: the change feed

/changes is the feed materializers and sync jobs consume. Same request as /read, plus types (default ["insert","erase"]).

curl -s https://api.capless.sh/v1/streams/events/changes -H "$H" -d '{"afterCursor":"cur_…","types":["insert","erase"]}'

{
  "changes": [
    { "type": "insert", "id": "ent_…", "cursor": "cur_…", "entry": { "user_id": "u3", "amount": 5 } },
    { "type": "erase", "id": "era_…", "cursor": "cur_…", "upToCursor": "cur_…", "filter": "entry.user_id == 'u1'",
      "status": "logically_applied", "affectedCount": 1, "affectedIds": ["ent_…"] }
  ],
  "blockedBy": null,
  "nextCursor": "cur_…", "snapshotCursor": "cur_…", "hasMore": false
}

• Inserts already covered by any accepted erasure are omitted: replaying from any cursor never resurrects erased records.
• Barrier: with erase in types, an erasure that is not yet resolved stops the page right before it (blockedBy is set, nextCursor is the boundary before the erasure). A consumer can therefore never advance its deletion checkpoint past an erasure it has not applied. Retry shortly; resolution is asynchronous.
• Resolved erasures carry the affected entry IDs inline (≤ 1000) or a paginated affectedIdsRef.
• Consumers that only want inserts pass "types": ["insert"] and are never blocked.
• Pagination never skips or repeats eligible items under a stable snapshot and erasure state.