# WebSocket stream Source: https://mostlyright.md/docs/recipes/websocket-stream/ Record a wss venue's frames first, then build a table from the recorded text relation. ## The question What price did the venue publish for each product, and when did each message arrive? ## The source | | | | --- | --- | | Venue | Coinbase Exchange market-data feed, `wss://ws-feed.exchange.coinbase.com` | | Members | `BTC-USD`, `ETH-USD` | | Format | JSON frames, written by a recorder into a fixed text relation | | Cadence | continuous while a subscription is open | | Rights | read under the venue's published market-data terms for its open feed, recorded as `terms_of_service_asserted` with the terms page digest as evidence | The feed is open, so it needs no credential. Its ticker messages carry a product identifier and a price, the smallest useful shape for this example. ## Record before you build A build never reaches the venue. A run reads the batches a recorder wrote before the run began. Four steps come in front of the ordinary build loop: ```sh mr-data stream document register --file connector.json --name coinbase --json mr-data stream registry create --digest DOCUMENT_DIGEST --json mr-data stream probe --document-digest DOCUMENT_DIGEST --registry REGISTRY_ID \ --member BTC-USD --json mr-data stream capture --document-digest DOCUMENT_DIGEST --registry REGISTRY_ID \ --member BTC-USD --member ETH-USD --seconds 300 --watch --json ``` 1. Register the connector document. It names the venue, the message type, the event time and the subscription frame. Registration prints the fingerprint the recipe names. `mr-data stream document show ID` reads one back. 2. Admit it into a registry, the reviewed set a recording may name. Pass every fingerprint the recording will use. 3. A probe connects briefly, reports which message types the document classified and which it did not, and records nothing. Read it with `mr-data stream status ID --kind probe`, and fix a document that classifies nothing before spending a capture window on it. 4. Capture a bounded window with `capture --seconds N`, where `--seconds` is at most 600, and `--max-events` and `--max-bytes` bound it further. `subscribe` instead keeps recording until `mr-data stream subscription stop ID`, checkpointing as it goes. `mr-data stream recordings --document-digest D` reports every batch a build could read for that venue, the bookmark each one carries, and when each expires. ## The recipe ```json title="recipe.json" { "dataset": { "name": "Coinbase Exchange ticker frames, BTC-USD and ETH-USD" }, "question": { "text": "What price did the venue publish for each product, and when did the message arrive?" }, "table": { "name": "ticker_frames", "description": "One row per recorded ticker message.", "grain": [ "event_id" ], "columns": [ { "name": "event_id", "type": "string", "nullable": false, "description": "The identity the recorder assigned this frame.", "presentation": { "chart": "none" } }, { "name": "received_at", "type": "timestamp", "nullable": false, "description": "When the recorder received the frame, in UTC.", "presentation": { "chart": "timeline", "bucket": "hour" } }, { "name": "product_id", "type": "string", "nullable": true, "description": "The market the message is about, as the venue names it.", "presentation": { "chart": "top_values" } }, { "name": "price", "type": "decimal", "nullable": true, "description": "The last trade price the venue published in this message.", "presentation": { "chart": "histogram", "bins": 10 } } ] }, "sources": [ { "name": "coinbase_ticker", "description": "The public Coinbase Exchange market-data feed, recorded frame by frame.", "source_class": "stream", "data_classification": "public", "locator": { "kind": "stream", "display_locator": "wss://ws-feed.exchange.coinbase.com" }, "rights_claim": { "claimed_basis": "terms_of_service_asserted", "claim_evidence_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "claim_note": "Read under the venue's published market-data terms for its open feed; the digest is of those terms as read on 12 September 2026." }, "connector": { "adapter_id": "stream.wss@1.0.0", "credential_mode": "none", "stream_connector_document_digest": "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", "parameters": [ { "name": "stream.members", "value": "BTC-USD,ETH-USD" } ] }, "limits": { "max_source_bytes": 268435456, "max_rows": 5000000 } } ], "transform": { "engine": "duckdb_sql", "steps": [ { "step_id": "ticker_messages", "sql": "select event_id, received_at, cast(json_extract(message, '$.product_id') as varchar) as product_id, cast(json_extract(message, '$.price') as varchar) as price from coinbase_ticker where message_type = 'ticker'", "description": "Keep the ticker frames and pull the two members this table needs out of the provider frame." }, { "step_id": "ticker_frames", "sql": "select event_id, cast(received_at as timestamptz) as received_at, replace(product_id, '\"', '') as product_id, try_cast(replace(price, '\"', '') as decimal(24, 8)) as price from ticker_messages", "description": "Cast the recorder's text. json_extract returns the JSON value, so a string member still carries its quotes; strip them before the numeric cast." } ] }, "checks": [ { "check_id": "one_row_per_frame", "kind": "key_uniqueness", "enforcement": "required", "columns": [ "event_id" ] }, { "check_id": "price_present", "kind": "null_ceiling", "enforcement": "required", "columns": [ "price" ], "max_null_ppm": 10000 }, { "check_id": "no_long_silence", "kind": "continuity", "enforcement": "advisory", "columns": [ "received_at" ], "max_gap_seconds": 300, "description": "A gap longer than five minutes is a window nobody was listening for, not a quiet market." }, { "check_id": "frames_present", "kind": "row_expectation", "enforcement": "required", "min_rows": 1 } ], "units": [], "timezone": "UTC" } ``` ## Part by part ### The source `source_class` is `stream` and `locator.kind` is `stream`, whose `display_locator` is the `wss://` endpoint. `credential_mode` is `none` for an open venue. A keyed venue uses `stream.wss@2.0.0` and a shaped credential enrolled with `mr-data keys set`, with the same document shape otherwise. `connector.stream_connector_document_digest` is the fingerprint step 1 printed. It binds this source to the connector document the capture ran under. `stream.members` names the members this source reads, comma separated. A source declares `stream.members` or `stream.series`, the rotating roster a subscription follows, never both. A source that names neither reads whatever recording binds to its connector document. The source declares no `window` and no `closed`. A bookmark continues a stream. ### The fixed relation A stream source reaches the SQL as the recorder's line, every column text, in this order. A connector that declares a member field adds a ninth column, `member`, after `event_time`. This recipe declares none, so the relation has these eight: | Column | Content | | --- | --- | | `epoch` | the connection this frame arrived on | | `event_id` | the identity the recorder assigned the frame | | `event_time` | the declared event time in the kind the connector declares, or the RFC 3339 arrival time when the recorder timed that frame by receipt | | `message` | the provider frame, canonical JSON text | | `message_class` | `snapshot`, `delta`, or whatever else the connector classifies | | `message_type` | the venue's own message type | | `received_at` | RFC 3339 UTC receipt time | | `sequence` | the frame's position within its connection | ### Transform Read the frame with the engine's JSON functions: ```sql select event_id, received_at, cast(json_extract(message, '$.product_id') as varchar) as product_id, cast(json_extract(message, '$.price') as varchar) as price from coinbase_ticker where message_type = 'ticker' ``` `json_extract` returns the JSON value, so a string member still carries its quotes. The second step strips them before the numeric cast: ```sql select event_id, cast(received_at as timestamptz) as received_at, replace(product_id, '"', '') as product_id, try_cast(replace(price, '"', '') as decimal(24, 8)) as price from ticker_messages ``` Run a sample and read `mr-data peek` before trusting that cast. The venue decides what goes in a member, and a `try_cast` returning null for every row looks like a quiet market. ### `event_time` holds two kinds of value, row by row A connector whose `event_time_field` is null times every frame by receipt. One whose `event_time_fallback` is `received_at` times only the frames the venue did not stamp. In one relation a row stamped by the venue carries the declared kind, say epoch milliseconds. The row beside it carries an RFC 3339 arrival time. A transform that casts the column blind gets nulls for half of it and no error. Read `received_at` when the question wants arrival time. Guard the cast on the shape when the question wants the venue's own time: ```sql select case when event_time like '%-%' then cast(event_time as timestamp) else to_timestamp(cast(event_time as bigint) / 1000) end as stamped_at from coinbase_ticker ``` A connector that declares an `event_time_field` and leaves `event_time_fallback` at `refuse`, the default, never mixes the two. Every frame carries the declared kind, or the capture refused it. ### Checks `key_uniqueness` on `event_id` is the grain. The `continuity` check on `received_at` with `max_gap_seconds: 300` is `advisory`. A gap longer than five minutes is a window nobody was listening for rather than a quiet market, and the run records it without ending. ### Units This document declares none. A currency is not in the unit vocabulary, and the vocabulary refuses a code outside its subset, naming the construct that put it there. A column with no physical unit stays out of `units`. ## Register, run, inspect Use the real table for one full build. For a bounded test first, follow the [scratch-table workflow](/docs/build/run-and-inspect/#test-before-the-full-build). A successful sample on the real table can publish truncated data. Stop if the forecast refuses; size the recipe before running. ```sh mr-data dataset create --name "Coinbase Exchange ticker frames" --json mr-data recipe recipe.json --json mr-data recipe explain RECIPE_ID --json mr-data recipe forecast RECIPE_ID --digest RECIPE_DIGEST --json mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --full --json mr-data watch RUN_ID --json mr-data peek RUN_ID --json mr-data query RUN_ID "select product_id, count(*) from run_table group by product_id" --json mr-data checks RUN_ID --json mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --refresh --json ``` `run` has no duration flag and will not grow one. A build reads recorded batches and stops at their end. The recorder fixed those bytes at capture time, not at build time. ## What the run reports The receipt names the recordings the run read and the bookmark it reached. Each run records that bookmark, and the next run continues from it. A refresh of a recorded stream rewrites no part. It keeps every part of the previous version and adds this run's. The table declares no `materialization` for a stream source, and the engine reads none. Such a refresh reads only the rows that arrived after the position the previous version committed at. ## Traps A missed window is a permanent hole. Recordings last thirty days, and a refresh continues from the bookmark the last build reached. Neither can recover messages that were never recorded. Report an unrecorded period as a gap in the series rather than promising a backfill, because there is nothing to fetch. Keep the subscription running where continuity matters. A refresh of a stream table needs a backend that writes a v2 manifest for a version composed at a recorder's checkpoint. Against one without it, every refresh of every stream table fails `PREDECESSOR_UNAVAILABLE` permanently. A full build does not converge either, because the next checkpoint writes a v1 manifest over it. When no recording exists, the build fails even if the recipe is valid. Run `mr-data stream recordings --document-digest D` and check what exists and when it expires. A keyed venue needs a shaped credential, stored by part: ```sh mr-data keys set VENUE_KEY --part key_id=ABC123 --part private_key_pem=@signing.pem --json mr-data stream capture --document-digest D --registry R --member M --seconds 300 \ --secret VENUE_KEY --json ``` `--part NAME=@PATH` reads that part from disk. The venue's document declares the parts its credential shape requires. The command names any set of parts that does not match. No command flag takes a secret as an argument. A probe that found the key wrong has done its job. It reports an `auth_outcome`: `authenticated`, `auth_rejected`, `auth_unconfirmed`, `clock_skew_suspected` and more. Act on that outcome.