diff --git a/src/api/tiled/README.md b/src/api/tiled/README.md index 8fc4c52e..165f74f7 100644 --- a/src/api/tiled/README.md +++ b/src/api/tiled/README.md @@ -24,7 +24,9 @@ import { getTiledSearch, setDefaultTiledUrl, setGlobalTiledApiKey } from '@/api/ setDefaultTiledUrl('http://localhost:8000/api/v1'); // include the version segment setGlobalTiledApiKey('…'); -const runs = await getTiledSearch('', { searchFilters: { specs: { include: ['BlueskyRun'], exclude: [] } } }); +const runs = await getTiledSearch('', { + searchFilters: { specs: { include: ['BlueskyRun'], exclude: [] } }, +}); console.log(runs.meta.count); ``` @@ -38,7 +40,13 @@ function RunBrowser() { if (runs.isPending) return

loading…

; if (runs.isError) return

{runs.error.message}

; - return ; + return ( + + ); } ``` @@ -83,14 +91,14 @@ call it in `beforeEach` so tests do not leak configuration into one another. ### Authentication -| What | How | -| --- | --- | -| API key in a header | default; `Authorization: ApiKey ` (the casing the package sends) | -| Spec casing | `setGlobalTiledApiKeyScheme('Apikey')` | -| API key in the query | `setGlobalTiledApiKeyLocation('query')` → `?api_key=` | -| Bearer token | `setDefaultBearerToken(jwt)`; takes precedence over the API key | -| No credentials, one call | `options.apiKey: null` — distinct from omitting it, which inherits | -| Refresh on 401 | automatic, single-flight, then one retry — but only when the call did not choose its own credentials and went to the configured server (see below). `/auth/session/refresh`, falling back to `/auth/refresh` on a 404 | +| What | How | +| ------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| API key in a header | default; `Authorization: ApiKey ` (the casing the package sends) | +| Spec casing | `setGlobalTiledApiKeyScheme('Apikey')` | +| API key in the query | `setGlobalTiledApiKeyLocation('query')` → `?api_key=` | +| Bearer token | `setDefaultBearerToken(jwt)`; takes precedence over the API key | +| No credentials, one call | `options.apiKey: null` — distinct from omitting it, which inherits | +| Refresh on 401 | automatic, single-flight, then one retry — but only when the call did not choose its own credentials and went to the configured server (see below). `/auth/session/refresh`, falling back to `/auth/refresh` on a 404 | The key is read **at request time** by a built-in interceptor, which is why `setApiKey` takes effect immediately. @@ -152,33 +160,33 @@ clearInterceptors(); // removes only YOUR interceptors The built-in auth and refresh handlers are tracked separately and survive `clearInterceptors()`. `setAxiosClient(next)` re-installs everything — built-ins first, then yours in registration order. -Axios runs request interceptors last-registered-first, so yours sees the config *before* +Axios runs request interceptors last-registered-first, so yours sees the config _before_ `Authorization` is attached. ## Four wire-format corrections These were found by probing a live Tiled 0.2.15b1, not by reading the spec — the spec says a -parameter's *type*, not how a list is serialised, and Tiled uses two different conventions. Each one +parameter's _type_, not how a list is serialised, and Tiled uses two different conventions. Each one failed quietly under `@blueskyproject/tiled`. **1. Three filter names were wrong.** FastAPI drops an unknown query parameter rather than rejecting it, so these filters simply never applied and the search came back unfiltered — which reads as "the filter matched everything", not as a bug. -| Filter | Package sent | Server expects | -| --- | --- | --- | -| `keysFilter` | `filter[keys][condition][keys][]` | `filter[keys_filter][condition][keys]` | -| `keyPresent` | `filter[key_present][condition][…]` | `filter[keypresent][condition][…]` | +| Filter | Package sent | Server expects | +| ------------ | ----------------------------------- | ------------------------------------------ | +| `keysFilter` | `filter[keys][condition][keys][]` | `filter[keys_filter][condition][keys]` | +| `keyPresent` | `filter[key_present][condition][…]` | `filter[keypresent][condition][…]` | | `accessBlob` | `filter[access_blob][condition][…]` | `filter[access_blob_filter][condition][…]` | **2. Axios brackets array parameters by default.** `column[]=a&column[]=b` where FastAPI reads -`column=a&column=b`. So `column` and `field` selections silently returned *everything*, and `fields` +`column=a&column=b`. So `column` and `field` selections silently returned _everything_, and `fields` never narrowed a response. Fixed with `paramsSerializer: { indexes: null }` on every request. **3. Four filters want one JSON array, not repeated keys.** `keys_filter.keys`, `in.value`, `notin.value` and `specs.include`/`exclude` travel as `filter[in][condition][value]=["count"]`. Sent as repeated keys, `in` matches nothing and -`keys_filter` answers **500**. `specs` additionally needs *both* `include` and `exclude` present — +`keys_filter` answers **500**. `specs` additionally needs _both_ `include` and `exclude` present — sending one alone is a 500, which is why `TiledSpecsFilter` requires both. **4. Filter values are JSON-encoded for you.** Six filters — `eq`, `noteq`, `comparison`, @@ -186,7 +194,9 @@ sending one alone is a 500, which is why `TiledSpecsFilter` requires both. `xas_scan` requires sending `"xas_scan"`, quotes included. Pass the value itself: ```ts -useTiledSearchQuery('', { searchFilters: { contains: { key: 'start.plan_name', value: 'xas_scan' } } }); +useTiledSearchQuery('', { + searchFilters: { contains: { key: 'start.plan_name', value: 'xas_scan' } }, +}); // on the wire: filter[contains][condition][value]="xas_scan" ``` @@ -203,11 +213,11 @@ useTiledSomethingQuery(...endpointArgs, queryOptions?, requestOptions?); useTiledSomethingMutation(mutationOptions?, requestOptions?); ``` -| position | what it is | -| --- | --- | -| the endpoint's own arguments | a path, a filter, a format — named and typed, so hover tells you what is required | -| TanStack options | `FinchQueryOptions` for queries, `FinchMutationOptions` for mutations | -| `requestOptions` | `baseUrl`, `apiKey`, `initialPath`, `pathMode`, `signal`, `client`, `headers`, `query`, `axiosConfig` | +| position | what it is | +| ---------------------------- | ----------------------------------------------------------------------------------------------------- | +| the endpoint's own arguments | a path, a filter, a format — named and typed, so hover tells you what is required | +| TanStack options | `FinchQueryOptions` for queries, `FinchMutationOptions` for mutations | +| `requestOptions` | `baseUrl`, `apiKey`, `initialPath`, `pathMode`, `signal`, `client`, `headers`, `query`, `axiosConfig` | **`requestOptions` is always last**, and means exactly the same thing on every Finch backend — where this one call goes and who it is. The queue-server hooks use the identical order; the convention is @@ -231,20 +241,21 @@ endpoint types. ### What is covered -| Group | Queries | Mutations | -| --- | --- | --- | -| Search | `useTiledSearchQuery` + 6 filter conveniences, `useTiledDistinctQuery` | — | -| Metadata | `useTiledMetadataQuery` | create, update, patch, delete | -| Arrays | 4 formats, `useTiledArrayBlockQuery`, `useTiledArrayImagePath` | put full, put block, patch full | -| Ragged | `useTiledRaggedFullQuery` | put full, put block, patch full | -| Tables | 5 JSON reads, `useTiledTableFullAsQuery` (CSV/parquet/arrow/…), 2 POST reads | put partition, patch partition, put full | -| Containers / nodes | full reads + POST variants | put node full | -| Awkward | full, buffers, POST buffers | put full | -| Management | revisions, assets | register, put data source, delete revision, close stream | -| Webhooks | list, history | register, delete | -| Server info | about, health, UI settings, metrics | — | -| Auth | whoami | login, logout, API key create/revoke, session refresh/revoke | -| Graph of links | entity, entities, link, links, namespaces, raw `useTiledGraphQLQuery` | entity / link create, update, delete; namespace upsert, delete; raw `useTiledGraphQLMutation` | +| Group | Queries | Mutations | +| ------------------ | ---------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------- | +| Search | `useTiledSearchQuery` + 6 filter conveniences, `useTiledDistinctQuery` | — | +| Metadata | `useTiledMetadataQuery` | create, update, patch, delete | +| Arrays | 4 formats, `useTiledArrayBlockQuery`, `useTiledArrayImagePath` | put full, put block, patch full | +| Ragged | `useTiledRaggedFullQuery` | put full, put block, patch full | +| Tables | 5 JSON reads, `useTiledTableFullAsQuery` (CSV/parquet/arrow/…), 2 POST reads | put partition, patch partition, put full | +| Containers / nodes | full reads + POST variants | put node full | +| Awkward | full, buffers, POST buffers | put full | +| Management | revisions, assets | register, put data source, delete revision, close stream | +| Webhooks | list, history | register, delete | +| Server info | about, health, UI settings, metrics | — | +| Auth | whoami | login, logout, API key create/revoke, session refresh/revoke | +| Graph of links | entity, entities, link, links, namespaces, raw `useTiledGraphQLQuery` | entity / link create, update, delete; namespace upsert, delete; raw `useTiledGraphQLMutation` | +| Streaming | — (websocket subscriptions, not queries; see Streaming below) | — | ### `useTiledDistinctQuery` is the one to know about @@ -287,14 +298,18 @@ const create = useTiledCreateNodeMutation(); const write = useTiledPutArrayFullMutation(); await create.mutateAsync({ - parentPath: '', // the CONTAINER to create in, not the new node's path + parentPath: '', // the CONTAINER to create in, not the new node's path body: { - id: 'processed', // the new node's key; omit for a server-assigned uuid + id: 'processed', // the new node's key; omit for a server-assigned uuid structure_family: 'array', metadata: {}, specs: [], access_blob: {}, - data_sources: [{ /* … structure describing what you are about to write … */ }], + data_sources: [ + { + /* … structure describing what you are about to write … */ + }, + ], }, }); @@ -346,7 +361,7 @@ Mutations invalidate named bundles automatically, awaited before `mutateAsync` r decide the map: - a **metadata write** refreshes that node and the searches that could have matched on what changed; -- a **data write** refreshes the data *and* the metadata, because a write can change a structure — +- a **data write** refreshes the data _and_ the metadata, because a write can change a structure — `patchArrayFull` with `extend: true` grows the shape, and a cached structure saying otherwise is what the downsampling maths reads; - a **create, delete or register** uses the `structure` bundle, which also covers the parent @@ -358,8 +373,8 @@ auth mutation invalidates `all`. ```ts const invalidate = useTiledInvalidate(); -await invalidate.roots('search'); // one resource -await invalidate.bundles('data'); // every structure family's payloads +await invalidate.roots('search'); // one resource +await invalidate.bundles('data'); // every structure family's payloads await invalidate.all(); ``` @@ -391,7 +406,7 @@ injected through `TiledApiProvider`. 2. otherwise the module-level default client. Setting `tiledApiUrl` / `tiledApiKey` on `FinchConfigProvider` is enough: those values are applied to -the default client *and* carried on every request, so even the first fetch of the first render uses +the default client _and_ carried on every request, so even the first fetch of the first render uses the configured server. Absent Finch config the client's own configuration stands. An **injected client is never redirected** — it is the caller's explicit choice, so Finch config is ignored for that subtree. @@ -430,7 +445,7 @@ For the `` **viewer component** and its CSS only — `src/components/Tile That component keeps its own internal client and its own singleton, so `setGlobalTiledApiKey` here does **not** configure it. `Tiled.tsx` passes the URL and key as props from Finch config, which is the -arrangement to keep. Login tokens *are* shared, through the `localStorage` keys above. +arrangement to keep. Login tokens _are_ shared, through the `localStorage` keys above. ## Completeness is checked, not claimed @@ -447,8 +462,8 @@ used a zarr library. ## Graph of links -Tiled's **experimental** entity/link graph (upstream #1387): *entities* — a project, a sample, a -dataset — optionally bound to a catalog node, and *links* `subject —predicate→ object` between them, +Tiled's **experimental** entity/link graph (upstream #1387): _entities_ — a project, a sample, a +dataset — optionally bound to a catalog node, and _links_ `subject —predicate→ object` between them, with CURIE namespaces for predicates. It is **GraphQL, not REST**, served at `{origin}/api/graphql` (outside `/api/v1`, like `/healthz`), and only on a catalog-backed server. @@ -469,11 +484,18 @@ import { await upsertTiledGraphNamespace('schema', 'https://schema.org/'); await upsertTiledGraphNamespace('prov', 'http://www.w3.org/ns/prov#'); -const project = await createTiledGraphEntity({ entityType: 'project', name: 'battery-cathodes-2026' }); +const project = await createTiledGraphEntity({ + entityType: 'project', + name: 'battery-cathodes-2026', +}); const sample = await createTiledGraphEntity({ entityType: 'sample', name: 'S001' }); const run = await bindTiledGraphEntity('raw/3f2c9a1e', { entityType: 'dataset' }); // a Bluesky run -await createTiledGraphLink({ subjectId: sample.id, predicate: 'schema:isPartOf', objectId: project.id }); +await createTiledGraphLink({ + subjectId: sample.id, + predicate: 'schema:isPartOf', + objectId: project.id, +}); await createTiledGraphLink({ subjectId: run.id, predicate: 'prov:used', objectId: sample.id }); ``` @@ -557,18 +579,91 @@ Deliberately not here yet: traversal helpers, auto-pagination and batch writes. around something upstream is likely to change (no counts, no bulk mutation, no uniqueness), so they wait for the API to settle and for a Finch screen that needs them. -## Not here yet +## Streaming (websockets) + +Tiled pushes live updates over `ws(s)://…/api/v1/stream/single/{path}`. The route is not in +`openapi.json`, so this layer is hand-written and lives in `client/websocket/`, with React hooks on +top. It is a **stream, not a query**: nothing here goes into the TanStack cache. The intended +pattern is that HTTP stays the source of truth, and a socket event tells you a cached query is stale: + +```tsx +const invalidate = useTiledInvalidate(); + +useTiledContainerSubscription('', { + onChildCreated: (m) => { + if (m.specs.some((s) => s.name === 'BlueskyRun')) invalidate.roots('search'); + }, + onChildMetadataUpdated: (m) => m.metadata.stop && invalidate.roots('search', 'metadata'), +}); + +useTiledTableSubscription(`${uid}/primary/internal`, { + onData: (update) => chart.append(update.payload), // { column: values[] } +}); -**Websockets.** Three seams are reserved so they land additively: +useTiledArraySubscription(`${uid}/primary/det_image`, { + onData: (update) => draw(update.payload, update.shape, update.offset), +}); +``` -- `apiKeyLocation: 'query'` is implemented already, because a browser websocket handshake cannot set - headers and query-string auth is the only mode that will work. The queue server learned this the - hard way; [its README](../qServer/README.md) documents the whole matrix. -- `closeStream` and the spec's `EventType` enum (`container-child-created`, - `container-child-metadata-updated`, `stream-closed`) are the streaming vocabulary. Webhooks already - model delivery; a socket transport will reuse those types rather than inventing parallel ones. -- Query roots are keyed by resource, not transport, so a socket that pushes an update invalidates - through the same `useTiledInvalidate()` surface. +Outside React, `new TiledWebSocketClient({ baseUrl, apiKey }).subscribe(path)` returns a +`TiledSubscription` with `connect()`, `disconnect()`, `reconnect()` and `onMessage` / `onStatus` / +`onError` (each returns an unsubscribe function). The hooks use the same URL, `initialPath` and +credentials as the HTTP hooks, so `FinchConfigProvider` is all the configuration needed. + +| hook | callbacks | state it keeps (re-renders) | +| ------------------------------- | ------------------------------------------------------ | ---------------------------------------------------------------------------------- | +| `useTiledSubscription` | `onMessage`, `onStatus`, `onError` | `lastMessage`/`sequence` per message unless `trackLastMessage: false` | +| `useTiledContainerSubscription` | `onSchema`, `onChildCreated`, `onChildMetadataUpdated` | status only | +| `useTiledTableSubscription` | `onSchema`, `onData` | `schema`; `latest` with `trackLatest`; `rows` only with `accumulate` (+ `maxRows`) | +| `useTiledArraySubscription` | `onSchema`, `onData`, `onRef` | `schema`; `latest`/`shape` only with `trackLatest`, never history | + +Callbacks are the preferred path for high-frequency data. Nothing accumulates unless you ask. + +What was verified against a live Tiled 0.2.15 server, and why the code looks the way it does: + +- **Subscriptions are per node and do not nest.** A container reports only its **direct** children + being created or having their metadata changed. Rows appended to `run/primary/internal` reach a + subscriber of that table, never a subscriber of `run` or of the root. +- **Auth is a first frame, not the URL.** Tiled reads a websocket API key only from the + `Authorization` header, which a browser cannot set, so `apiKeyLocation: 'query'` does **not** work + here, despite what this README used to say. When a key or token is configured the socket sends + `{"type": "auth", "api_key": …}` (or `access_token`) on open. A server that requires auth waits for + it. A server with anonymous reads never reads it, so it is harmless there. Close 4001/4003 means a + bad credential and is not retried. +- **The root container needs a trailing slash**: `/stream/single/`. The slashless form is a 403. +- **`start` is inclusive.** Reconnects ask for `lastSequence + 1`, so the gap replays exactly, as far + back as the server's streaming cache retains it. Duplicates are dropped by sequence. If the + server's counter goes backwards between connections (a restart with an in-memory cache), the + subscription starts a new epoch instead of silently swallowing every later event. +- **Every connection starts with a `*-schema` message** carrying no sequence and no data. In + particular, `table-schema` describes the columns, not the rows already there. Load existing data + over HTTP, or pass `start: 0` to replay what is retained. +- **A server close with code 1000 ends the subscription** ("Producer ended stream"); it is not + retried. +- **Tables stream before they persist.** If persisting fails and the writer retries, each retry is a + fresh `table-data` message with a new sequence. De-duplication cannot catch that; it shows up as + repeated rows. +- **Array payloads cannot be suppressed.** Every `array-data` message carries the written data in + full, so there is deliberately no `includePayload` option. The lightweight alternative, + `array-ref` (a shape plus a `uri` to fetch the slice over HTTP), is only emitted for data + registered through `put_data_source`, not for ordinary PUT/PATCH writes. For a large detector, + that means either registering external files, or accepting the bandwidth. +- **The server must be configured for it.** Without a `streaming_cache` in Tiled's config, every + stream connection is rejected with a 500. `streaming_cache: {uri: memory}` is enough for a single + process; use Redis when Tiled is scaled out or replay should survive a restart. +- **Origin is not checked**, so CORS settings do not affect websockets. + +Not implemented yet: MsgPack envelopes (binary frames are dropped and counted, reported as `parse` +errors), Arrow decoding of `table-schema.arrow_schema`, automatic fetching of `array-ref` data, +multiplexing several subscriptions over one socket, and Bluesky-specific helpers such as a run +subscription. Those belong above this layer. + +`TiledWebSocket.live.test.ts` exercises all of this against a real server. It is skipped unless +`TILED_LIVE_URL` is set: + +```sh +TILED_LIVE_URL=http://localhost:8000/api/v1 TILED_LIVE_API_KEY=… npx vitest run TiledWebSocket.live +``` ## Known duplication diff --git a/src/api/tiled/SKILLS.md b/src/api/tiled/SKILLS.md index d8b425ac..f102f041 100644 --- a/src/api/tiled/SKILLS.md +++ b/src/api/tiled/SKILLS.md @@ -24,6 +24,12 @@ client/ searchParams.ts filter + option encoding (the four wire-format corrections live here) arraySlicing.ts downsampling maths, ported from upstream TiledArrayApi tokenStorage.ts injectable login-token store; localStorage keys shared with + websocket/ + types.ts stream message union, connection status/error, WebSocketLike + buildWebSocketUrl.ts http(s) API URL -> ws(s)://…/stream/single/{path}?envelope_format=… + parseMessage.ts JSON envelope -> typed message or null (unknown types dropped) + TiledSubscription.ts one node's socket: first-frame auth, backoff, sequence/replay, listeners + TiledWebSocketClient.ts shared credentials -> subscribe(path) endpoints/ readEndpoints.ts registry descriptors: info, search, metadata GET, data reads, assets @@ -75,6 +81,9 @@ hooks/ infoHooks.ts 5 queries authHooks.ts 1 query + 6 mutations graphHooks.ts graph of links — 7 queries + 10 mutations (incl. raw GraphQL, node binding) + useTiledSubscription.ts generic stream hook (NOT a query; no cache involvement) + useTiled{Container,Table,Array}Subscription.ts typed wrappers over it + internal/useTiledSocketConfig.ts URL / initialPath / credentials, resolved like the HTTP hooks typeTests.ts compile-time assertions (tsconfig excludes src/testing, so they live here) index.ts barrel ``` @@ -98,7 +107,7 @@ The spec is the source of truth and the coverage test enforces it. Five steps, i 2. **`client/TiledApiClient.ts`** — add the method. Use the `get`/`post`/`put`/`patch`/`del` funnels, or `sendBinary` for a binary body. Node paths go through `this.dataPath(...)`, never `buildPath`. 3. **`client/facade.ts`** — add the free function. -4. **`runtime/clientLike.ts`** — add the method name to *both* the `Pick` and +4. **`runtime/clientLike.ts`** — add the method name to _both_ the `Pick` and `TILED_CLIENT_LIKE_METHODS` (the `satisfies` makes one without the other a compile error). 5. **`endpoints/*.ts`** — add the descriptor, with the spec's `operationId`. @@ -177,6 +186,11 @@ without one fails CI. The two zarr URL builders are the only entries mapped to ` are described; an explicit `null` survives an update while an omitted field is absent; 401 refreshes. - hooks: id/document guards, one `graph` root, graph writes and `deleteNode` refresh it. +`TiledWebSocket.test.ts` / `TiledSubscriptionHooks.test.tsx`: URL building, one parser fixture per +message type (copied from a live server), lifecycle and close codes, backoff, sequence de-duplication +and epoch reset, hook mount/unmount/path change, and the no-accumulation defaults. +`TiledWebSocket.live.test.ts` runs end to end when `TILED_LIVE_URL` is set. + `TiledClient.test.ts`: path encoding, origin derivation, downsampling maths (including the non-array-structure guard), format resolution, JSON-seq parsing, search option mapping, error normalisation. @@ -190,7 +204,7 @@ normalisation. - **GraphQL fails with HTTP 200.** A resolver error arrives as `{ data, errors }` with status 200; `graphql()` rejects whenever `errors` is non-empty. Never branch on status for a graph call. - **The graph never answers 401.** Graph access control filters results instead of refusing: with an - access policy, an unauthenticated read returns empty lists; on a server *without* one (e.g. + access policy, an unauthenticated read returns empty lists; on a server _without_ one (e.g. single-user `--api-key`), it returns everything — verified against tiled@849c984a, where REST `search` answered 401 for the same anonymous caller. Never treat a successful graph read as proof of authentication, and treat an empty graph as "check the key" before "nothing there". @@ -204,7 +218,7 @@ normalisation. wrong is a silent no-op or a 500. Both forms are in `client/searchParams.ts` with the evidence. - **`paramsSerializer: { indexes: null }` is load-bearing.** Axios otherwise writes `column[]=`, which FastAPI drops — so a column selection silently returns every column. Set on the instance - *and* per request, so an adopted axios client behaves too. + _and_ per request, so an adopted axios client behaves too. - **Six filter values are JSON, not text.** `eq`, `noteq`, `comparison`, `contains`, `in`, `notin`. Callers pass real values; `searchParams.ts` encodes. Do not encode `fulltext`/`regex`/`like`/ `lookup`/`structureFamily` — those are plain strings. @@ -252,6 +266,17 @@ normalisation. - **`@blueskyproject/tiled` is still installed, for the `` viewer only.** Nothing in this folder may import it. The viewer has its own client and singleton; only `localStorage` tokens are shared. +- **Websocket auth is a first frame.** The server reads `api_key` from the `Authorization` header + only, so `apiKeyLocation: 'query'` does nothing for sockets. Do not "fix" this by putting the key in + the URL. +- **`/stream/single/` needs its trailing slash for the root**; `buildTiledWebSocketUrl` keeps it. +- **`start` is inclusive**; reconnect with `lastSequence + 1`. The first sequenced frame of a new + connection may legitimately go _backwards_ (server counter reset), which is why it is not dropped. +- **Stream hooks must not re-render per message by default.** The typed wrappers pass + `trackLastMessage: false`; anything that keeps state (`accumulate`, `trackLatest`) is opt-in. +- **jsdom's `WebSocket` is broken under vitest** (it trips over its own `Event` class). Unit tests use + `FakeSocket` from `src/testing/tests/tiledStreamFixtures.ts`; the live test injects the `ws` + package through `socketFactory`. - **`src/components/Tiled/types/tempTypes.ts` is a stale duplicate** of the node/structure types, kept for the viewer's prop types. It still types `data_sources` as `string | null`, which is wrong. Do not add call sites; import from `@/api/tiled`. diff --git a/src/api/tiled/client/websocket/TiledSubscription.ts b/src/api/tiled/client/websocket/TiledSubscription.ts new file mode 100644 index 00000000..6ca29783 --- /dev/null +++ b/src/api/tiled/client/websocket/TiledSubscription.ts @@ -0,0 +1,364 @@ +import { buildTiledWebSocketUrl } from './buildWebSocketUrl'; +import { isTiledSchemaMessage, parseTiledStreamMessage } from './parseMessage'; +import type { + TiledConnectionStatus, + TiledSocketFactory, + TiledStreamMessage, + TiledSubscriptionConfig, + TiledSubscriptionError, + Unsubscribe, + WebSocketLike, +} from './types'; + +/** Close codes Tiled's `/stream/single` route uses (see `tiled/server/router.py`). */ +export const TILED_WS_CLOSE = { + /** A normal close. From the server it means the producer ended the stream. */ + NORMAL: 1000, + /** The first frame was not valid JSON, or not `{"type": "auth", …}`. */ + BAD_AUTH_MESSAGE: 4001, + /** The first-frame credentials were rejected. */ + AUTH_FAILED: 4003, +} as const; + +const JITTER = 0.2; + +/** + * One live subscription to one Tiled node. Framework-independent: no React and no TanStack Query. + * + * ```ts + * const sub = new TiledSubscription({ baseUrl: 'http://localhost:8000/api/v1', path: 'run/primary/internal' }); + * const off = sub.onMessage((m) => { if (m.type === 'table-data') plot(m.payload); }); + * sub.connect(); + * // later + * off(); + * sub.disconnect(); + * ``` + * + * Behaviour worth knowing: + * + * - **Auth is a first frame.** When an API key or access token is configured, the socket sends + * `{"type": "auth", …}` as soon as it opens. A server that requires auth waits for that frame. A + * server that allows anonymous reads never reads it, so sending it there does no harm. Close 4001 + * or 4003 means the credentials were rejected, and the subscription does not retry with the same + * bad key. + * - **Reconnects resume at `lastSequence + 1`.** Tiled's `start` is inclusive, so this replays exactly + * the gap, as far back as the server's streaming cache retains it. A message whose sequence is at or + * below the last one delivered is dropped. The one exception is the first sequenced message on a new + * connection: if that one is at or below `lastSequence`, the server's counter was reset (a restart + * with an in-memory cache, or a stream that closed and reopened), so it starts a new epoch and is + * delivered. Without this, every later event would be silently swallowed. + * - **The schema message is re-sent on every connection** and is always delivered, so a consumer can + * treat it as "(re)connected, here is the current structure". + * - **A server close with 1000 ends the subscription.** Tiled sends it when the producer closes the + * stream, so reconnecting would only wait on a stream that is finished. + * - **No outbound traffic** after the auth frame. The server ignores client frames and has no + * application-level ping. + */ +export class TiledSubscription { + readonly path: string; + + private readonly config: TiledSubscriptionConfig; + private readonly openSocket: TiledSocketFactory; + + private socket: WebSocketLike | null = null; + private status: TiledConnectionStatus = 'idle'; + private lastSequence: number | undefined; + private closedByUser = false; + private attempt = 0; + private dropped = 0; + private reconnectTimer: ReturnType | null = null; + /** Whether the current connection has delivered a sequenced message yet (see the epoch rule). */ + private sawSequenceThisConnection = false; + + private readonly messageListeners = new Set<(message: TiledStreamMessage) => void>(); + private readonly statusListeners = new Set<(status: TiledConnectionStatus) => void>(); + private readonly errorListeners = new Set<(error: TiledSubscriptionError) => void>(); + + constructor(config: TiledSubscriptionConfig) { + this.config = config; + this.path = config.path; + this.openSocket = + config.socketFactory ?? + ((url: string) => new WebSocket(url) as unknown as WebSocketLike); + } + + // #region public API + + /** Open the socket. Does nothing while a socket is already open or connecting. */ + connect(): void { + this.closedByUser = false; + if (this.socket) return; + this.clearReconnectTimer(); + this.open(); + } + + /** Close for good. No reconnect follows until `connect()` is called again. */ + disconnect(): void { + this.closedByUser = true; + this.clearReconnectTimer(); + this.detachSocket(TILED_WS_CLOSE.NORMAL); + this.setStatus('closed'); + } + + /** Drop the current socket and connect again now, resetting the backoff. Replay still applies. */ + reconnect(): void { + this.closedByUser = false; + this.clearReconnectTimer(); + this.attempt = 0; + this.detachSocket(TILED_WS_CLOSE.NORMAL); + this.open(); + } + + onMessage(listener: (message: TiledStreamMessage) => void): Unsubscribe { + this.messageListeners.add(listener); + return () => { + this.messageListeners.delete(listener); + }; + } + + /** Calling this immediately replays the current status to the new listener. */ + onStatus(listener: (status: TiledConnectionStatus) => void): Unsubscribe { + this.statusListeners.add(listener); + listener(this.status); + return () => { + this.statusListeners.delete(listener); + }; + } + + onError(listener: (error: TiledSubscriptionError) => void): Unsubscribe { + this.errorListeners.add(listener); + return () => { + this.errorListeners.delete(listener); + }; + } + + getStatus(): TiledConnectionStatus { + return this.status; + } + + /** The highest sequence delivered so far, or `undefined` before the first sequenced message. */ + getLastSequence(): number | undefined { + return this.lastSequence; + } + + /** Frames dropped because they were not a recognised message. Duplicates are not counted. */ + getDroppedCount(): number { + return this.dropped; + } + + // #endregion + + // #region connection + + private open(): void { + this.setStatus(this.attempt > 0 ? 'reconnecting' : 'connecting'); + this.sawSequenceThisConnection = false; + + const url = buildTiledWebSocketUrl(this.config.baseUrl, this.config.path, { + envelopeFormat: this.config.envelopeFormat, + start: this.lastSequence !== undefined ? this.lastSequence + 1 : this.config.start, + }); + + let ws: WebSocketLike; + try { + ws = this.openSocket(url); + } catch (error) { + this.emitError({ kind: 'transport', message: describeError(error) }); + this.scheduleReconnect(); + return; + } + this.socket = ws; + + ws.onopen = () => { + const auth = this.authFrame(); + if (!auth) return; + try { + ws.send(JSON.stringify(auth)); + } catch (error) { + this.emitError({ kind: 'auth', message: describeError(error) }); + } + }; + + ws.onmessage = (event) => { + if (this.socket !== ws) return; + if (this.status !== 'connected') { + this.attempt = 0; + this.setStatus('connected'); + } + const message = parseTiledStreamMessage(event.data); + if (!message) { + this.dropped += 1; + this.emitError({ + kind: 'parse', + message: + typeof event.data === 'string' + ? 'Dropped a frame that is not a known Tiled stream message.' + : 'Dropped a binary frame; only JSON envelopes are decoded.', + }); + return; + } + if (!this.acceptSequence(message)) return; + this.dispatch(message); + }; + + // Browsers give no detail on `error`, and a `close` with the code always follows, so the + // error is reported once, from onclose. + ws.onerror = () => {}; + + ws.onclose = (event) => { + if (this.socket !== ws) return; + this.socket = null; + if (this.closedByUser) { + this.setStatus('closed'); + return; + } + + const code = event?.code; + if (code === TILED_WS_CLOSE.BAD_AUTH_MESSAGE || code === TILED_WS_CLOSE.AUTH_FAILED) { + this.emitError({ + kind: 'auth', + code, + message: + code === TILED_WS_CLOSE.AUTH_FAILED + ? 'Tiled rejected the credentials (close 4003).' + : `Tiled expected an auth frame (close 4001${event.reason ? `: ${event.reason}` : ''}).`, + }); + this.setStatus('error'); + return; + } + + if (code === TILED_WS_CLOSE.NORMAL) { + // The producer ended the stream. A future stream on this node restarts its + // sequence at 1, so forget ours. + this.lastSequence = undefined; + this.setStatus('closed'); + return; + } + + this.emitError({ + kind: 'transport', + code, + message: `Connection closed unexpectedly (code ${code ?? 'unknown'}).`, + }); + this.scheduleReconnect(); + }; + } + + private scheduleReconnect(): void { + if (this.closedByUser || this.config.reconnect === false) { + this.setStatus('error'); + return; + } + const maxAttempts = this.config.maxReconnectAttempts ?? Number.POSITIVE_INFINITY; + if (this.attempt >= maxAttempts) { + this.emitError({ + kind: 'reconnect-exhausted', + message: `Giving up after ${this.attempt} reconnect attempts.`, + }); + this.setStatus('error'); + return; + } + + const initial = this.config.reconnectDelay ?? 1000; + const ceiling = this.config.maxReconnectDelay ?? 30_000; + const base = Math.min(ceiling, initial * 2 ** this.attempt); + const delay = Math.max(0, Math.round(base + base * JITTER * (Math.random() * 2 - 1))); + + this.attempt += 1; + this.setStatus('reconnecting'); + this.clearReconnectTimer(); + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + if (!this.closedByUser) this.open(); + }, delay); + } + + /** Close the current socket without letting its late events touch this subscription. */ + private detachSocket(code: number): void { + const current = this.socket; + if (!current) return; + this.socket = null; + current.onopen = null; + current.onmessage = null; + current.onerror = null; + current.onclose = null; + try { + current.close(code); + } catch { + // Already closing. + } + } + + private clearReconnectTimer(): void { + if (this.reconnectTimer !== null) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + } + + private authFrame(): Record | null { + if (this.config.accessToken) return { type: 'auth', access_token: this.config.accessToken }; + if (this.config.apiKey) return { type: 'auth', api_key: this.config.apiKey }; + return null; + } + + // #endregion + + // #region dispatch + + /** The sequence rules from the class docs. Returns whether to deliver the message. */ + private acceptSequence(message: TiledStreamMessage): boolean { + if (isTiledSchemaMessage(message)) return true; + const { sequence } = message; + const firstOnConnection = !this.sawSequenceThisConnection; + this.sawSequenceThisConnection = true; + + if (this.lastSequence !== undefined && sequence <= this.lastSequence) { + if (!firstOnConnection) return false; + // The server's counter went backwards between connections: a new epoch. + } + this.lastSequence = sequence; + return true; + } + + private dispatch(message: TiledStreamMessage): void { + for (const listener of this.messageListeners) { + try { + listener(message); + } catch (error) { + console.error(`[tiled-ws ${this.path || '/'}] message listener threw:`, error); + } + } + } + + private setStatus(next: TiledConnectionStatus): void { + if (this.status === next) return; + this.status = next; + for (const listener of this.statusListeners) { + try { + listener(next); + } catch (error) { + console.error(`[tiled-ws ${this.path || '/'}] status listener threw:`, error); + } + } + } + + private emitError(error: TiledSubscriptionError): void { + for (const listener of this.errorListeners) { + try { + listener(error); + } catch (listenerError) { + console.error( + `[tiled-ws ${this.path || '/'}] error listener threw:`, + listenerError, + ); + } + } + } + + // #endregion +} + +function describeError(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} diff --git a/src/api/tiled/client/websocket/TiledWebSocketClient.ts b/src/api/tiled/client/websocket/TiledWebSocketClient.ts new file mode 100644 index 00000000..f1bf2736 --- /dev/null +++ b/src/api/tiled/client/websocket/TiledWebSocketClient.ts @@ -0,0 +1,28 @@ +import { TiledSubscription } from './TiledSubscription'; +import type { TiledSubscriptionOptions, TiledWebSocketClientOptions } from './types'; + +/** + * Holds one server's URL and credentials and creates subscriptions against it. + * + * ```ts + * const client = new TiledWebSocketClient({ baseUrl: 'http://localhost:8000/api/v1', apiKey: 'secret' }); + * const sub = client.subscribe('run-id/primary/internal'); + * sub.onMessage((message) => console.log(message)); + * sub.connect(); + * ``` + * + * Each subscription owns its own socket, because Tiled's route serves one node per connection. The + * client does not multiplex subscriptions or track them; disconnecting each one is up to the caller. + */ +export class TiledWebSocketClient { + private readonly options: TiledWebSocketClientOptions; + + constructor(options: TiledWebSocketClientOptions) { + this.options = options; + } + + /** Create a subscription to `path`. It does not connect until `connect()` is called. */ + subscribe(path: string, options: TiledSubscriptionOptions = {}): TiledSubscription { + return new TiledSubscription({ ...this.options, ...options, path }); + } +} diff --git a/src/api/tiled/client/websocket/buildWebSocketUrl.ts b/src/api/tiled/client/websocket/buildWebSocketUrl.ts new file mode 100644 index 00000000..90eaab0e --- /dev/null +++ b/src/api/tiled/client/websocket/buildWebSocketUrl.ts @@ -0,0 +1,35 @@ +import { encodeTiledPath, normalizeTiledBaseUrl } from '../urlUtils'; +import type { TiledSubscriptionOptions } from './types'; + +/** The streaming route, relative to the API root (`…/api/v1`). Not in `openapi.json`. */ +export const TILED_STREAM_PATH = '/stream/single/'; + +/** + * Build the websocket URL for one node from the HTTP API URL. + * + * `http://host:8000/api/v1` + `'run/primary'` → `ws://host:8000/api/v1/stream/single/run/primary?envelope_format=json`. + * + * - The trailing slash on `/stream/single/` is required. The server routes the root container only as + * `/stream/single/`, and answers the slashless form with a 403. + * - Each path segment is encoded separately (`encodeTiledPath`), so `/` keeps separating segments while + * `?`, `#` and spaces inside a key are escaped. + * - Credentials never go in the URL. Tiled reads a websocket API key only from the `Authorization` + * header, which a browser cannot set, so `TiledSubscription` sends them as the first frame instead. + */ +export function buildTiledWebSocketUrl( + apiUrl: string, + path: string, + options: Pick = {}, +): string { + const url = new URL(normalizeTiledBaseUrl(apiUrl)); + if (url.protocol === 'https:') url.protocol = 'wss:'; + else if (url.protocol === 'http:') url.protocol = 'ws:'; + + url.pathname = `${url.pathname.replace(/\/+$/, '')}${TILED_STREAM_PATH}${encodeTiledPath(path)}`; + url.search = ''; + url.hash = ''; + url.searchParams.set('envelope_format', options.envelopeFormat ?? 'json'); + if (options.start !== undefined) url.searchParams.set('start', String(options.start)); + + return url.toString(); +} diff --git a/src/api/tiled/client/websocket/index.ts b/src/api/tiled/client/websocket/index.ts new file mode 100644 index 00000000..5e918f12 --- /dev/null +++ b/src/api/tiled/client/websocket/index.ts @@ -0,0 +1,8 @@ +/** + * Tiled's streaming websocket, without React. The hooks in `../../hooks` sit on top of this. + */ +export { TiledWebSocketClient } from './TiledWebSocketClient'; +export { TiledSubscription, TILED_WS_CLOSE } from './TiledSubscription'; +export { buildTiledWebSocketUrl, TILED_STREAM_PATH } from './buildWebSocketUrl'; +export { parseTiledStreamMessage, isTiledSchemaMessage } from './parseMessage'; +export type * from './types'; diff --git a/src/api/tiled/client/websocket/parseMessage.ts b/src/api/tiled/client/websocket/parseMessage.ts new file mode 100644 index 00000000..aebb018e --- /dev/null +++ b/src/api/tiled/client/websocket/parseMessage.ts @@ -0,0 +1,60 @@ +import type { TiledStreamMessage, TiledStreamMessageType } from './types'; + +const SCHEMA_TYPES = new Set([ + 'container-schema', + 'table-schema', + 'array-schema', + 'ragged-schema', +]); + +const SEQUENCED_TYPES = new Set([ + 'container-child-created', + 'container-child-metadata-updated', + 'table-data', + 'array-data', + 'array-ref', + 'ragged-data', +]); + +/** + * Parse one JSON envelope into a typed message, or return `null` for a frame to drop. + * + * Only the discriminant is checked, plus a numeric `sequence` on the types that carry one; the rest of + * the message is trusted as sent. Deep validation would be too slow for large array payloads on this + * hot path. + * + * Returns `null` for: + * - binary frames, which MsgPack envelopes will need a decoder for; + * - invalid JSON; + * - an unknown `type`, which is what a newer server's additions look like. + */ +export function parseTiledStreamMessage(raw: unknown): TiledStreamMessage | null { + if (typeof raw !== 'string') return null; + + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return null; + } + if (typeof parsed !== 'object' || parsed === null) return null; + + const type = (parsed as { type?: unknown }).type; + if (typeof type !== 'string') return null; + + if (SCHEMA_TYPES.has(type as TiledStreamMessageType)) { + return parsed as TiledStreamMessage; + } + if (SEQUENCED_TYPES.has(type as TiledStreamMessageType)) { + const sequence = (parsed as { sequence?: unknown }).sequence; + return typeof sequence === 'number' ? (parsed as TiledStreamMessage) : null; + } + return null; +} + +/** Whether a message is the schema message sent first on every connection. */ +export function isTiledSchemaMessage( + message: TiledStreamMessage, +): message is Extract { + return SCHEMA_TYPES.has(message.type); +} diff --git a/src/api/tiled/client/websocket/types.ts b/src/api/tiled/client/websocket/types.ts new file mode 100644 index 00000000..2977d530 --- /dev/null +++ b/src/api/tiled/client/websocket/types.ts @@ -0,0 +1,271 @@ +import type { Spec } from '../../types/generatedAliases'; + +/** + * Wire types for Tiled's streaming websocket, `GET /api/v1/stream/single/{path}`. + * + * Every shape here was checked against a live server (Tiled 0.2.15) and its source + * (`tiled/catalog/adapter.py`, `tiled/server/core.py`), not just the docs. The route is not in + * `openapi.json`, so nothing here is generated. + * + * Each connection first sends a `*-schema` message, which has no `sequence`. After that, every message + * carries a per-node `sequence` that starts at 1 and increases by one. A container's sequence counts + * child events, so it is unrelated to any child's own sequence. + */ + +// #region messages + +/** Sent first on every connection to a container. Carries no sequence. */ +export interface ContainerSchemaMessage { + type: 'container-schema'; + version: number; +} + +/** A direct child was created. Nested descendants do **not** produce this on the parent. */ +export interface ContainerChildCreatedMessage { + type: 'container-child-created'; + sequence: number; + /** Server-local ISO timestamp with no timezone, e.g. `2026-10-06T18:56:21.633871`. */ + timestamp: string; + key: string; + structure_family: string; + specs: Spec[]; + metadata: Record; + data_sources: unknown[]; + access_blob: Record; +} + +/** A direct child's metadata or specs were replaced or patched. */ +export interface ContainerChildMetadataUpdatedMessage { + type: 'container-child-metadata-updated'; + sequence: number; + timestamp: string; + key: string; + specs: Spec[]; + metadata: Record; + /** Missing when the update was made with `drop_revision`. */ + revision_number?: number; +} + +/** Sent first on every connection to a table. Carries no sequence and no rows. */ +export interface TableSchemaMessage { + type: 'table-schema'; + version: number; + /** + * The table's Arrow schema. Over JSON it arrives as a + * `data:application/vnd.apache.arrow.file;base64,…` URL holding an Arrow IPC message. Finch does + * not decode it, so it is typed as an opaque string. + */ + arrow_schema: string; +} + +/** + * Rows written to a table. + * + * In JSON envelopes the server transcodes the Arrow body to a **columnar** object, `{column: values[]}`, + * hence the default `T`. + * + * Tiled streams the write *before* it persists it. If persisting fails and the writer retries, each + * retry arrives as a fresh `table-data` with a new sequence, so sequence de-duplication cannot merge + * them. + */ +export interface TableDataMessage> { + type: 'table-data'; + sequence: number; + timestamp: string; + /** Media type of the original upload, e.g. `application/vnd.apache.arrow.file`. */ + mimetype: string; + /** Media type of `payload` as sent: `application/json` in JSON envelopes. */ + 'content-type'?: string; + partition: number | null; + append: boolean; + payload: T; +} + +/** The fields of Tiled's `BuiltinDtype`. Structured dtypes carry extra keys. */ +export interface TiledStreamDataType { + endianness?: string; + kind?: string; + itemsize?: number; + dt_units?: string | null; + [key: string]: unknown; +} + +/** Sent first on every connection to an array. Carries no sequence. */ +export interface ArraySchemaMessage { + type: 'array-schema'; + version: number; + data_type: TiledStreamDataType; +} + +/** + * Array data that was written, sent in full, payload included. + * + * - `shape` is the shape of `payload`, not of the whole array. + * - `offset` is where a patch starts. It can have fewer entries than `shape` has dimensions; for + * example a `[1, 3]` patch at row 2 arrives with `offset: [2]`. + * - `offset` and `block` are both `null` for a full overwrite. + * + * The server cannot leave the payload out of this message, so subscribing to a large detector array + * costs the full bandwidth. + */ +export interface ArrayDataMessage { + type: 'array-data'; + sequence: number; + timestamp: string; + mimetype: string; + 'content-type'?: string; + shape: number[]; + offset: number[] | null; + block: number[] | null; + payload: T; +} + +/** + * Says that new array data was registered, without sending the data. + * + * Tiled sends this only when a data source is attached with `put_data_source`, which is how externally + * written files (HDF5, TIFF, …) are registered. An ordinary PUT/PATCH of array data produces + * `array-data` instead. To get the bytes, fetch `uri` over HTTP when the UI needs them. + */ +export interface ArrayRefMessage { + type: 'array-ref'; + sequence: number; + timestamp: string; + data_source: Record; + patch: { offset: number[]; shape: number[]; [key: string]: unknown } | null; + shape: number[]; + /** An `/array/full/{path}?slice=…` URL covering the new region. */ + uri: string; +} + +/** Sent first on every connection to a ragged array. */ +export interface RaggedSchemaMessage { + type: 'ragged-schema'; + version: number; + [key: string]: unknown; +} + +/** Ragged data that was written. Over JSON the payload is nested lists. */ +export interface RaggedDataMessage { + type: 'ragged-data'; + sequence: number; + timestamp: string; + mimetype: string; + 'content-type'?: string; + payload: T; + [key: string]: unknown; +} + +/** Every message a Tiled stream can deliver. Narrow it on `type`. */ +export type TiledStreamMessage = + | ContainerSchemaMessage + | ContainerChildCreatedMessage + | ContainerChildMetadataUpdatedMessage + | TableSchemaMessage + | TableDataMessage + | ArraySchemaMessage + | ArrayDataMessage + | ArrayRefMessage + | RaggedSchemaMessage + | RaggedDataMessage; + +export type TiledStreamMessageType = TiledStreamMessage['type']; + +/** The message sent first on every (re)connection. */ +export type TiledSchemaMessage = Extract; + +/** Messages that carry a `sequence`, i.e. everything except the schema message. */ +export type TiledSequencedMessage = Exclude; + +// #endregion + +// #region connection + +/** + * - `idle`: created or disabled, with no socket opened yet. + * - `connecting`: the handshake is in flight. + * - `connected`: the first frame (the schema) has arrived. + * - `reconnecting`: the connection dropped and a retry is scheduled. + * - `closed`: closed on purpose, either by `disconnect()` or by the producer ending the stream. + * - `error`: a failure that will not be retried (bad credentials, reconnects exhausted). + */ +export type TiledConnectionStatus = + | 'idle' + | 'connecting' + | 'connected' + | 'reconnecting' + | 'closed' + | 'error'; + +/** + * - `auth`: the server rejected the credentials (close 4001 or 4003). Not retried. + * - `transport`: the socket errored or dropped. Retried when reconnect is on. + * - `parse`: a frame was not a known message. Dropped, and the stream continues. + * - `reconnect-exhausted`: `maxReconnectAttempts` ran out. + */ +export type TiledSubscriptionErrorKind = 'auth' | 'transport' | 'parse' | 'reconnect-exhausted'; + +export interface TiledSubscriptionError { + kind: TiledSubscriptionErrorKind; + message: string; + /** Websocket close code, when the error came from a close frame. */ + code?: number; +} + +export type Unsubscribe = () => void; + +/** + * The part of the `WebSocket` API this module uses, so tests can inject a fake. + * + * This is the same shape as the queue server's `WebSocketLike`. It is declared again here so the two + * API modules stay independent. + */ +export interface WebSocketLike { + readyState: number; + send(data: string): void; + close(code?: number, reason?: string): void; + onopen: ((event: unknown) => void) | null; + onmessage: ((event: { data: unknown }) => void) | null; + onerror: ((event: unknown) => void) | null; + onclose: ((event: { code: number; reason?: string }) => void) | null; +} + +export type TiledSocketFactory = (url: string) => WebSocketLike; + +/** Credentials and transport shared by every subscription a `TiledWebSocketClient` creates. */ +export interface TiledWebSocketClientOptions { + /** The Tiled API URL, **including `/api/v1`**, as for the HTTP client. */ + baseUrl: string; + apiKey?: string | null; + accessToken?: string | null; + /** Defaults to `new WebSocket(url)`. */ + socketFactory?: TiledSocketFactory; +} + +/** Options for one subscription. */ +export interface TiledSubscriptionOptions { + /** + * Replay from this sequence number (inclusive), as far back as the server's streaming cache still + * holds data. `0` replays everything that is retained. Leave it out to get only new events. + */ + start?: number; + /** Only `'json'` is decoded today. `'msgpack'` is reserved, and its binary frames are dropped. */ + envelopeFormat?: 'json' | 'msgpack'; + /** Reconnect after an unexpected drop. Default `true`. */ + reconnect?: boolean; + /** First retry delay in ms. Default 1000. */ + reconnectDelay?: number; + /** Ceiling for the exponential backoff in ms. Default 30000. */ + maxReconnectDelay?: number; + /** Default unlimited. */ + maxReconnectAttempts?: number; +} + +/** Everything a `TiledSubscription` needs: one node, plus the shared credentials and transport. */ +export interface TiledSubscriptionConfig + extends TiledWebSocketClientOptions, TiledSubscriptionOptions { + /** Node path relative to the API root. `''` is the root container. */ + path: string; +} + +// #endregion diff --git a/src/api/tiled/hooks/index.ts b/src/api/tiled/hooks/index.ts index ca584199..ae5f512d 100644 --- a/src/api/tiled/hooks/index.ts +++ b/src/api/tiled/hooks/index.ts @@ -220,3 +220,31 @@ export type { } from './graphHooks'; // #endregion + +// #region Streaming +// Websocket subscriptions. These are streams, not queries; they invalidate the cache through +// callbacks rather than living in it. + +export { useTiledSubscription } from './useTiledSubscription'; +export type { + UseTiledSubscriptionOptions, + UseTiledSubscriptionResult, +} from './useTiledSubscription'; +export { useTiledContainerSubscription } from './useTiledContainerSubscription'; +export type { + UseTiledContainerSubscriptionOptions, + UseTiledContainerSubscriptionResult, +} from './useTiledContainerSubscription'; +export { useTiledTableSubscription, appendColumnarRows } from './useTiledTableSubscription'; +export type { + TiledColumnarRows, + UseTiledTableSubscriptionOptions, + UseTiledTableSubscriptionResult, +} from './useTiledTableSubscription'; +export { useTiledArraySubscription } from './useTiledArraySubscription'; +export type { + UseTiledArraySubscriptionOptions, + UseTiledArraySubscriptionResult, +} from './useTiledArraySubscription'; + +// #endregion diff --git a/src/api/tiled/hooks/internal/useTiledSocketConfig.ts b/src/api/tiled/hooks/internal/useTiledSocketConfig.ts new file mode 100644 index 00000000..075fbbab --- /dev/null +++ b/src/api/tiled/hooks/internal/useTiledSocketConfig.ts @@ -0,0 +1,58 @@ +import { useMemo } from 'react'; +import { getDefaultTiledApiClient } from '../../client/defaultClient'; +import { resolveTiledPath } from '../../client/urlUtils'; +import { INJECTED_CLIENT_SCOPE } from '../queryKeys'; +import { useTiledClient } from '../useTiledClient'; + +export interface TiledSocketConfig { + /** `undefined` when an injected stub client has no URL, so there is nothing to connect to. */ + baseUrl: string | undefined; + /** `path` with the client's `initialPath` applied, the same way HTTP requests are resolved. */ + path: string; + apiKey: string | undefined; + accessToken: string | undefined; +} + +/** + * Where a streaming hook should connect, and with which credentials, using the same resolution as the + * HTTP hooks: + * + * - With a client injected through `TiledApiProvider`, its URL, initial path and API key are used. + * - Otherwise `FinchConfigProvider`'s `tiledApiUrl` / `tiledApiKey` are used, falling back to the + * default client. + * - A login session's access token is read from the default client, and only when no client is + * injected. `TiledClientLike` does not expose tokens, and widening it for this is not worth it. + */ +export function useTiledSocketConfig(path: string): TiledSocketConfig { + const { client, source, requestDefaults, scope } = useTiledClient(); + + let accessToken: string | undefined; + if (source === 'default') { + const fallback = getDefaultTiledApiClient(); + accessToken = + fallback.getBearerToken() ?? fallback.getStoredTokens()?.accessToken ?? undefined; + } + + const apiKey = requestDefaults.apiKey ?? readOptional(() => client.getApiKey()) ?? undefined; + const scopeUrl = scope.baseUrl === INJECTED_CLIENT_SCOPE ? undefined : scope.baseUrl; + const baseUrl = requestDefaults.baseUrl ?? scopeUrl; + const resolvedPath = resolveTiledPath(path, scope.initialPath); + + return useMemo( + () => ({ baseUrl, path: resolvedPath, apiKey, accessToken }), + [baseUrl, resolvedPath, apiKey, accessToken], + ); +} + +/** + * A partial injected stub has a throwing placeholder for each method it lacks (see + * `completeClientSurface`). For a socket, a missing credential getter just means "no credential", + * not a broken render. + */ +function readOptional(read: () => T): T | undefined { + try { + return read(); + } catch { + return undefined; + } +} diff --git a/src/api/tiled/hooks/useTiledArraySubscription.ts b/src/api/tiled/hooks/useTiledArraySubscription.ts new file mode 100644 index 00000000..8d4323eb --- /dev/null +++ b/src/api/tiled/hooks/useTiledArraySubscription.ts @@ -0,0 +1,98 @@ +import { useEffect, useState } from 'react'; +import type { + ArrayDataMessage, + ArrayRefMessage, + ArraySchemaMessage, +} from '../client/websocket/types'; +import { + useTiledSubscription, + type UseTiledSubscriptionOptions, + type UseTiledSubscriptionResult, +} from './useTiledSubscription'; + +export interface UseTiledArraySubscriptionOptions extends Omit< + UseTiledSubscriptionOptions, + 'onMessage' | 'trackLastMessage' +> { + onSchema?: (message: ArraySchemaMessage) => void; + /** Called for every `array-data` message, payload included. */ + onData?: (message: ArrayDataMessage) => void; + /** + * Called for every `array-ref` message: new data was registered and nothing was sent. Fetch + * `message.uri` over HTTP when the UI actually needs the bytes. + */ + onRef?: (message: ArrayRefMessage) => void; + /** + * Keep the newest message in `latest`, with its `shape`, re-rendering per message. Default `false`. + * Only one message is ever kept; earlier payloads are not. + */ + trackLatest?: boolean; +} + +export interface UseTiledArraySubscriptionResult extends Omit< + UseTiledSubscriptionResult, + 'lastMessage' +> { + /** The schema (dtype) from the most recent (re)connection. */ + schema?: ArraySchemaMessage; + /** Only populated with `trackLatest`. */ + latest?: ArrayDataMessage | ArrayRefMessage; + /** Shape of `latest`, i.e. of the region that was written. Only populated with `trackLatest`. */ + shape?: number[]; +} + +/** + * Stream writes to one array node, e.g. a detector image series. + * + * ```tsx + * useTiledArraySubscription(`${uid}/primary/det_image`, { + * onData: (update) => drawImage(update.payload, update.shape, update.offset), + * }); + * ``` + * + * Tiled sends the **full payload** of every `array-data` write, and there is no way to ask it not to. + * Ignoring the payload does not save bandwidth, so think twice before subscribing to a large detector. + * Only data registered through `put_data_source` arrives as the lightweight `array-ref`. + * + * No history is kept. Payloads go to `onData` and are not retained unless `trackLatest` is set, and + * even then only the newest one is kept. + */ +export function useTiledArraySubscription( + path: string, + options: UseTiledArraySubscriptionOptions = {}, +): UseTiledArraySubscriptionResult { + const { onSchema, onData, onRef, trackLatest = false, ...rest } = options; + + const [schema, setSchema] = useState(); + const [latest, setLatest] = useState | ArrayRefMessage>(); + + useEffect(() => { + setSchema(undefined); + setLatest(undefined); + }, [path, trackLatest]); + + const { lastMessage: _lastMessage, ...result } = useTiledSubscription(path, { + ...rest, + trackLastMessage: false, + onMessage(message) { + switch (message.type) { + case 'array-schema': + setSchema(message); + onSchema?.(message); + break; + case 'array-data': { + const data = message as ArrayDataMessage; + if (trackLatest) setLatest(data); + onData?.(data); + break; + } + case 'array-ref': + if (trackLatest) setLatest(message); + onRef?.(message); + break; + } + }, + }); + + return { ...result, schema, latest, shape: latest?.shape }; +} diff --git a/src/api/tiled/hooks/useTiledContainerSubscription.ts b/src/api/tiled/hooks/useTiledContainerSubscription.ts new file mode 100644 index 00000000..69d761c0 --- /dev/null +++ b/src/api/tiled/hooks/useTiledContainerSubscription.ts @@ -0,0 +1,67 @@ +import type { + ContainerChildCreatedMessage, + ContainerChildMetadataUpdatedMessage, + ContainerSchemaMessage, +} from '../client/websocket/types'; +import { + useTiledSubscription, + type UseTiledSubscriptionOptions, + type UseTiledSubscriptionResult, +} from './useTiledSubscription'; + +export interface UseTiledContainerSubscriptionOptions extends Omit< + UseTiledSubscriptionOptions, + 'onMessage' | 'trackLastMessage' +> { + /** Sent on every (re)connection, so it doubles as an "is live" signal. */ + onSchema?: (message: ContainerSchemaMessage) => void; + /** A direct child was created, e.g. a Bluesky run starting under the root. */ + onChildCreated?: (message: ContainerChildCreatedMessage) => void; + /** A direct child's metadata changed, e.g. a Bluesky run gaining `metadata.stop`. */ + onChildMetadataUpdated?: (message: ContainerChildMetadataUpdatedMessage) => void; +} + +export type UseTiledContainerSubscriptionResult = Omit; + +/** + * Watch a container for direct-child changes. + * + * ```tsx + * const invalidate = useTiledInvalidate(); + * useTiledContainerSubscription('', { + * onChildCreated: () => invalidate.roots('search'), + * onChildMetadataUpdated: ({ key, metadata }) => { + * if (metadata.stop) console.log('run finished', key); + * }, + * }); + * ``` + * + * Only **direct** children are reported. A table or array further down sends nothing to its ancestors, + * so subscribe to it with `useTiledTableSubscription` or `useTiledArraySubscription`. The hook never + * re-renders per message; react to events through the callbacks. + */ +export function useTiledContainerSubscription( + path: string, + options: UseTiledContainerSubscriptionOptions = {}, +): UseTiledContainerSubscriptionResult { + const { onSchema, onChildCreated, onChildMetadataUpdated, ...rest } = options; + + const { lastMessage: _lastMessage, ...result } = useTiledSubscription(path, { + ...rest, + trackLastMessage: false, + onMessage(message) { + switch (message.type) { + case 'container-schema': + onSchema?.(message); + break; + case 'container-child-created': + onChildCreated?.(message); + break; + case 'container-child-metadata-updated': + onChildMetadataUpdated?.(message); + break; + } + }, + }); + return result; +} diff --git a/src/api/tiled/hooks/useTiledSubscription.ts b/src/api/tiled/hooks/useTiledSubscription.ts new file mode 100644 index 00000000..a2988f07 --- /dev/null +++ b/src/api/tiled/hooks/useTiledSubscription.ts @@ -0,0 +1,179 @@ +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import { TiledSubscription } from '../client/websocket/TiledSubscription'; +import type { + TiledConnectionStatus, + TiledSocketFactory, + TiledStreamMessage, + TiledSubscriptionError, + TiledSubscriptionOptions, + WebSocketLike, +} from '../client/websocket/types'; +import { useTiledSocketConfig } from './internal/useTiledSocketConfig'; + +export interface UseTiledSubscriptionOptions extends TiledSubscriptionOptions { + /** `false` keeps the socket closed and leaves the hook `idle`. Default `true`. */ + enabled?: boolean; + /** Called for every delivered message. The latest function is always used; it need not be stable. */ + onMessage?: (message: TiledStreamMessage) => void; + onStatus?: (status: TiledConnectionStatus) => void; + onError?: (error: TiledSubscriptionError) => void; + /** + * Mirror every message into `lastMessage` and `sequence` state, re-rendering once per message. + * Default `true`. + * + * Turn this off for high-frequency streams and use `onMessage` instead. `sequence` is then only + * refreshed when the status changes. + */ + trackLastMessage?: boolean; + /** Override the socket constructor, mainly for tests. Read when the socket opens. */ + socketFactory?: TiledSocketFactory; +} + +export interface UseTiledSubscriptionResult { + status: TiledConnectionStatus; + lastMessage?: TiledStreamMessage; + /** The highest sequence delivered. See `trackLastMessage` for how often it updates. */ + sequence?: number; + /** The most recent error. Cleared on reconnect and when the subscription target changes. */ + error?: TiledSubscriptionError; + /** Close the socket. It stays closed until `reconnect()` is called or the inputs change. */ + disconnect(): void; + /** Reconnect now, resuming from the last sequence when the server can replay it. */ + reconnect(): void; +} + +/** + * Subscribe to one Tiled node's stream for the lifetime of a component. + * + * ```tsx + * const { status, lastMessage } = useTiledSubscription('run-id/primary/internal'); + * + * useTiledSubscription(path, { + * trackLastMessage: false, + * onMessage(message) { + * if (message.type === 'table-data') chart.append(message.payload); + * }, + * }); + * ``` + * + * Connects with the same URL, `initialPath` and credentials as the HTTP hooks (`FinchConfigProvider`, + * or a client injected through `TiledApiProvider`). A new socket is opened when the path, URL, + * credentials, `envelopeFormat`, `start`, reconnect options or `enabled` change, and closed on + * unmount. + * + * This is a stream, not a query, and it takes no part in the TanStack cache. To keep HTTP data fresh, + * invalidate it from a callback, for example with `useTiledInvalidate().roots('search')`. + * + * The typed wrappers `useTiledContainerSubscription`, `useTiledTableSubscription` and + * `useTiledArraySubscription` are usually the more convenient entry points. + */ +export function useTiledSubscription( + path: string, + options: UseTiledSubscriptionOptions = {}, +): UseTiledSubscriptionResult { + const { + enabled = true, + trackLastMessage = true, + start, + envelopeFormat, + reconnect: reconnectEnabled, + reconnectDelay, + maxReconnectDelay, + maxReconnectAttempts, + } = options; + const { baseUrl, path: resolvedPath, apiKey, accessToken } = useTiledSocketConfig(path); + + const [status, setStatus] = useState('idle'); + const [lastMessage, setLastMessage] = useState(); + const [sequence, setSequence] = useState(); + const [error, setError] = useState(); + + // Held in refs so an inline callback or factory does not tear the socket down every render. + const callbacksRef = useRef(options); + callbacksRef.current = options; + const trackRef = useRef(trackLastMessage); + trackRef.current = trackLastMessage; + + const subscriptionRef = useRef(null); + + useEffect(() => { + setLastMessage(undefined); + setSequence(undefined); + setError(undefined); + + if (!enabled || !baseUrl) { + setStatus('idle'); + return; + } + + const subscription = new TiledSubscription({ + baseUrl, + path: resolvedPath, + apiKey, + accessToken, + start, + envelopeFormat, + reconnect: reconnectEnabled, + reconnectDelay, + maxReconnectDelay, + maxReconnectAttempts, + socketFactory: (url) => { + const factory = callbacksRef.current.socketFactory; + return factory ? factory(url) : (new WebSocket(url) as unknown as WebSocketLike); + }, + }); + subscriptionRef.current = subscription; + + const unsubscribers = [ + subscription.onStatus((next) => { + setStatus(next); + setSequence(subscription.getLastSequence()); + callbacksRef.current.onStatus?.(next); + }), + subscription.onMessage((message) => { + if (trackRef.current) { + setLastMessage(message); + setSequence(subscription.getLastSequence()); + } + callbacksRef.current.onMessage?.(message); + }), + subscription.onError((socketError) => { + setError(socketError); + callbacksRef.current.onError?.(socketError); + }), + ]; + subscription.connect(); + + return () => { + for (const unsubscribe of unsubscribers) unsubscribe(); + subscription.disconnect(); + if (subscriptionRef.current === subscription) subscriptionRef.current = null; + }; + }, [ + enabled, + baseUrl, + resolvedPath, + apiKey, + accessToken, + start, + envelopeFormat, + reconnectEnabled, + reconnectDelay, + maxReconnectDelay, + maxReconnectAttempts, + ]); + + const disconnect = useCallback(() => { + subscriptionRef.current?.disconnect(); + }, []); + + const reconnect = useCallback(() => { + setError(undefined); + subscriptionRef.current?.reconnect(); + }, []); + + return useMemo( + () => ({ status, lastMessage, sequence, error, disconnect, reconnect }), + [status, lastMessage, sequence, error, disconnect, reconnect], + ); +} diff --git a/src/api/tiled/hooks/useTiledTableSubscription.ts b/src/api/tiled/hooks/useTiledTableSubscription.ts new file mode 100644 index 00000000..0cf5b85c --- /dev/null +++ b/src/api/tiled/hooks/useTiledTableSubscription.ts @@ -0,0 +1,139 @@ +import { useEffect, useState } from 'react'; +import type { TableDataMessage, TableSchemaMessage } from '../client/websocket/types'; +import { + useTiledSubscription, + type UseTiledSubscriptionOptions, + type UseTiledSubscriptionResult, +} from './useTiledSubscription'; + +/** Columnar rows, the shape Tiled's JSON envelope gives a table payload. */ +export type TiledColumnarRows = Record; + +export interface UseTiledTableSubscriptionOptions extends Omit< + UseTiledSubscriptionOptions, + 'onMessage' | 'trackLastMessage' +> { + onSchema?: (message: TableSchemaMessage) => void; + /** Called for every `table-data` message. This is the preferred path for live plots. */ + onData?: (message: TableDataMessage) => void; + /** + * Concatenate every payload into `rows`, re-rendering per message. Default `false`, because a long + * scan would otherwise grow without bound in React state. Assumes the columnar JSON payload. + * + * Every payload is appended, including a non-`append` partition overwrite; reconstructing a table + * from overwrites is not attempted. + */ + accumulate?: boolean; + /** With `accumulate`, keep only the newest this many rows. */ + maxRows?: number; + /** Keep the newest message in `latest`, re-rendering per message. Implied by `accumulate`. */ + trackLatest?: boolean; +} + +export interface UseTiledTableSubscriptionResult extends Omit< + UseTiledSubscriptionResult, + 'lastMessage' +> { + /** The schema from the most recent (re)connection. */ + schema?: TableSchemaMessage; + /** Only populated with `trackLatest` or `accumulate`. */ + latest?: TableDataMessage; + /** Only populated with `accumulate`. Rows sent before the subscription started are not included. */ + rows?: TiledColumnarRows; +} + +/** + * Stream rows appended to one table node, e.g. a Bluesky run's `primary/internal`. + * + * ```tsx + * useTiledTableSubscription(`${uid}/primary/internal`, { + * onData: (update) => chart.append(update.payload), + * }); + * + * const { rows } = useTiledTableSubscription(path, { accumulate: true, maxRows: 10_000 }); + * ``` + * + * The schema message describes the columns, not the rows already in the table. Fetch those once over + * HTTP (`useTiledTableFullAsJSONQuery`), or pass `start: 0` to replay what the streaming cache still + * holds. + */ +export function useTiledTableSubscription( + path: string, + options: UseTiledTableSubscriptionOptions = {}, +): UseTiledTableSubscriptionResult { + const { onSchema, onData, accumulate = false, maxRows, trackLatest = false, ...rest } = options; + const keepLatest = trackLatest || accumulate; + + const [schema, setSchema] = useState(); + const [latest, setLatest] = useState>(); + const [rows, setRows] = useState(); + + // A new target starts from nothing. + useEffect(() => { + setSchema(undefined); + setLatest(undefined); + setRows(undefined); + }, [path, accumulate]); + + const { lastMessage: _lastMessage, ...result } = useTiledSubscription(path, { + ...rest, + trackLastMessage: false, + onMessage(message) { + if (message.type === 'table-schema') { + setSchema(message); + onSchema?.(message); + } else if (message.type === 'table-data') { + const data = message as TableDataMessage; + if (keepLatest) setLatest(data); + if (accumulate) { + setRows((current) => + appendColumnarRows(current, data.payload as TiledColumnarRows, maxRows), + ); + } + onData?.(data); + } + }, + }); + + return { ...result, schema, latest, rows: accumulate ? (rows ?? {}) : undefined }; +} + +/** + * Append columnar `next` to `current`, keeping the newest `maxRows` rows. + * + * A column that appears part-way through is back-filled with `null`, and a column missing from `next` + * is padded with `null`, so every column always has the same length. + */ +export function appendColumnarRows( + current: TiledColumnarRows | undefined, + next: TiledColumnarRows | null | undefined, + maxRows?: number, +): TiledColumnarRows { + const base = current ?? {}; + if (!next || typeof next !== 'object') return base; + + const existingLength = columnLength(base); + const incomingLength = columnLength(next); + const merged: TiledColumnarRows = {}; + + for (const column of new Set([...Object.keys(base), ...Object.keys(next)])) { + const before = base[column] ?? new Array(existingLength).fill(null); + const added = Array.isArray(next[column]) + ? next[column] + : new Array(incomingLength).fill(null); + let values = before.concat(added); + if (maxRows !== undefined && values.length > maxRows) { + values = values.slice(values.length - maxRows); + } + merged[column] = values; + } + return merged; +} + +function columnLength(rows: TiledColumnarRows): number { + let length = 0; + for (const values of Object.values(rows)) { + if (Array.isArray(values)) length = Math.max(length, values.length); + } + return length; +} diff --git a/src/api/tiled/index.ts b/src/api/tiled/index.ts index 2d1846e8..b88d836f 100644 --- a/src/api/tiled/index.ts +++ b/src/api/tiled/index.ts @@ -122,6 +122,12 @@ export type { TiledTokenStorage, TiledStoredTokens } from './client/tokenStorage */ export * from './client/facade'; +/** + * The streaming websocket (`/stream/single/{path}`), without React. The `useTiled*Subscription` + * hooks above are built on it. + */ +export * from './client/websocket'; + /** Content negotiation: format names, their media types, and the JSON-sequence parser. */ export { TILED_FORMATS, resolveFormat, parseJsonSequence } from './client/formats'; export type { TiledFormatName, TiledFormatSpec, TiledFormatReturnMap } from './client/formats'; diff --git a/src/app/App.tsx b/src/app/App.tsx index a5d254e6..af4b1228 100644 --- a/src/app/App.tsx +++ b/src/app/App.tsx @@ -8,6 +8,7 @@ import Documentation from './pages/Documentation'; import Experiment from '@/components/Experiment/Experiment'; import TestTiled from '@/components/TiledTest/TestTiled'; +import TiledWebsocket from '@/components/TiledWebsocket/TiledWebsocket'; import TiledQueryPlayground from '@/components/TiledQueryPlayground/TiledQueryPlayground'; import FinchAppLayout from '@/components/FinchAppLayout/FinchAppLayout'; @@ -37,7 +38,7 @@ function App() { classNameContainer: 'bg-slate-50', }, { - // Three manual testbeds, as real nested routes rather than local state — so each has + // Four manual testbeds, as real nested routes rather than local state — so each has // its own url, back and forward work, and a reload keeps you where you were. path: '/test', label: 'Test', @@ -80,6 +81,16 @@ function App() { isBackgroundTransparent: false, classNameContainer: 'h-auto min-h-full bg-slate-50 p-4 text-slate-800', }, + { + // Tiled's streaming websocket: the subscription hooks, the raw + // TiledSubscription, replay and reconnects, with buttons that make the + // server emit events. Same light surface as the other Tiled harnesses. + path: 'tiled-websocket', + label: 'Tiled WebSocket', + element: , + isBackgroundTransparent: false, + classNameContainer: 'h-auto min-h-full bg-slate-50 p-4 text-slate-800', + }, ], }, { diff --git a/src/components/TiledWebsocket/ArrayPreview.tsx b/src/components/TiledWebsocket/ArrayPreview.tsx new file mode 100644 index 00000000..5284b384 --- /dev/null +++ b/src/components/TiledWebsocket/ArrayPreview.tsx @@ -0,0 +1,77 @@ +import { useEffect, useMemo, useRef } from 'react'; + +/** + * Draws a 1-D or 2-D numeric payload as a greyscale image, scaled to its own min and max. + * + * This is the "draw from the callback, don't keep history" pattern for detector data. Only the + * newest frame is ever held, and it is painted straight to a canvas. + */ +export default function ArrayPreview({ payload }: { payload: unknown }) { + const canvasRef = useRef(null); + const grid = useMemo(() => toGrid(payload), [payload]); + + useEffect(() => { + const canvas = canvasRef.current; + if (!canvas || !grid) return; + const rows = grid.length; + const cols = grid[0].length; + canvas.width = cols; + canvas.height = rows; + const context = canvas.getContext('2d'); + if (!context) return; + + let min = Infinity; + let max = -Infinity; + for (const row of grid) { + for (const value of row) { + if (value < min) min = value; + if (value > max) max = value; + } + } + const span = max - min || 1; + const image = context.createImageData(cols, rows); + grid.forEach((row, y) => + row.forEach((value, x) => { + const level = Math.round(((value - min) / span) * 255); + const index = (y * cols + x) * 4; + image.data[index] = level; + image.data[index + 1] = level; + image.data[index + 2] = level; + image.data[index + 3] = 255; + }), + ); + context.putImageData(image, 0, 0); + }, [grid]); + + if (!grid) { + return ( +

+ No preview: the payload is not a 1-D or 2-D array of numbers. +

+ ); + } + // Square pixels, fitted into 384×192 CSS px, so a one-row patch shows as a strip, not a block. + const rows = grid.length; + const cols = grid[0].length; + const scale = Math.min(384 / cols, 192 / rows, 12); + return ( + + ); +} + +function toGrid(payload: unknown): number[][] | null { + if (!Array.isArray(payload) || payload.length === 0) return null; + const rows = Array.isArray(payload[0]) ? (payload as unknown[][]) : [payload]; + const width = rows[0].length; + if (width === 0) return null; + for (const row of rows) { + if (!Array.isArray(row) || row.length !== width) return null; + if (!row.every((value) => typeof value === 'number')) return null; + } + return rows as number[][]; +} diff --git a/src/components/TiledWebsocket/StreamLogView.tsx b/src/components/TiledWebsocket/StreamLogView.tsx new file mode 100644 index 00000000..984d5b87 --- /dev/null +++ b/src/components/TiledWebsocket/StreamLogView.tsx @@ -0,0 +1,105 @@ +import { useState } from 'react'; +import JsonResultViewer from '../devtools/JsonResultViewer'; +import type { StreamLog, StreamLogEntry } from './useStreamLog'; + +const KIND_COLORS: Record = { + status: 'text-slate-500', + error: 'text-rose-600', +}; + +/** + * A subscription's event log: per-type counters, a filter, a pause switch, and expandable rows. + * + * Pausing freezes what is on screen. Messages keep arriving and being counted, so you can read a + * fast stream without it scrolling away. + */ +export default function StreamLogView({ log }: { log: StreamLog }) { + const [filter, setFilter] = useState('all'); + const [paused, setPaused] = useState(null); + const [openId, setOpenId] = useState(null); + + const source = paused ?? log.entries; + const shown = filter === 'all' ? source : source.filter((entry) => entry.kind === filter); + const kinds = Object.keys(log.counts).sort(); + + return ( +
+
+ + {log.total.toLocaleString()} messages + + {log.rate.toFixed(1)}/s + {kinds.map((kind) => ( + + {kind} ×{log.counts[kind]} + + ))} + + + +
+ +
    + {shown.length === 0 &&
  1. (nothing yet)
    2. } + {shown.map((entry) => ( +
  2. + + {openId === entry.id && entry.detail !== undefined && ( +
    + +
    + )} +
    3. + ))} +
+
+ ); +} diff --git a/src/components/TiledWebsocket/SubscriptionPanel.tsx b/src/components/TiledWebsocket/SubscriptionPanel.tsx new file mode 100644 index 00000000..69189d22 --- /dev/null +++ b/src/components/TiledWebsocket/SubscriptionPanel.tsx @@ -0,0 +1,748 @@ +import { useEffect, useMemo, useState } from 'react'; +import { + TiledWebSocketClient, + buildTiledWebSocketUrl, + resolveTiledPath, + useTiledArraySubscription, + useTiledContainerSubscription, + useTiledInvalidate, + useTiledSubscription, + useTiledTableSubscription, +} from '@/api/tiled'; +import type { + ArrayDataMessage, + ArrayRefMessage, + ContainerChildCreatedMessage, + ContainerChildMetadataUpdatedMessage, + TiledConnectionStatus, + TiledSocketFactory, + TiledSubscription, + TiledSubscriptionError, + TiledSubscriptionOptions, +} from '@/api/tiled'; +import type { TiledConnectionConfig } from '../devtools/TiledConnectionBar'; +import JsonResultViewer from '../devtools/JsonResultViewer'; +import ArrayPreview from './ArrayPreview'; +import { MODE_LABELS, modeForFamily, type SubscriptionMode } from './modes'; +import StreamLogView from './StreamLogView'; +import { useStreamLog, type StreamLog } from './useStreamLog'; + +// #region options + +/** The form's fields, as typed. Numbers stay strings until Subscribe, so a field can be empty. */ +interface DraftOptions { + start: string; + reconnect: boolean; + reconnectDelay: string; + maxReconnectDelay: string; + maxReconnectAttempts: string; + accumulate: boolean; + maxRows: string; + trackLatest: boolean; + invalidateOnChange: boolean; +} + +const DEFAULT_DRAFT: DraftOptions = { + start: '', + reconnect: true, + reconnectDelay: '1000', + maxReconnectDelay: '30000', + maxReconnectAttempts: '', + accumulate: false, + maxRows: '500', + trackLatest: true, + invalidateOnChange: true, +}; + +interface CommittedOptions { + subscription: TiledSubscriptionOptions; + accumulate: boolean; + maxRows?: number; + trackLatest: boolean; + invalidateOnChange: boolean; +} + +function commit(draft: DraftOptions): CommittedOptions { + const number = (value: string) => { + const parsed = Number(value); + return value.trim() === '' || Number.isNaN(parsed) ? undefined : parsed; + }; + return { + subscription: { + start: number(draft.start), + reconnect: draft.reconnect, + reconnectDelay: number(draft.reconnectDelay), + maxReconnectDelay: number(draft.maxReconnectDelay), + maxReconnectAttempts: number(draft.maxReconnectAttempts), + }, + accumulate: draft.accumulate, + maxRows: number(draft.maxRows), + trackLatest: draft.trackLatest, + invalidateOnChange: draft.invalidateOnChange, + }; +} + +// #endregion + +export interface SubscriptionPanelProps { + initialPath: string; + initialMode: SubscriptionMode; + connection: TiledConnectionConfig; + onRemove: () => void; + /** Open another panel, e.g. on a child a container subscription just reported. */ + onSubscribeChild: (path: string, mode: SubscriptionMode) => void; + /** Overrides the browser `WebSocket`; for tests. */ + socketFactory?: TiledSocketFactory; +} + +const inputClass = + 'w-full rounded border border-slate-300 bg-white px-2 py-1 font-mono text-xs text-slate-900 disabled:bg-slate-100'; +const buttonClass = + 'rounded border border-slate-300 px-2 py-1 text-xs hover:bg-slate-100 disabled:opacity-50'; + +/** + * One subscription: pick a hook, set its options, subscribe, and watch what arrives. + * + * Options are staged and only take effect on Subscribe, the same way the connection bar stages its + * edits, because changing `start` or a reconnect option opens a new socket. + */ +export default function SubscriptionPanel({ + initialPath, + initialMode, + connection, + onRemove, + onSubscribeChild, + socketFactory, +}: SubscriptionPanelProps) { + const [mode, setMode] = useState(initialMode); + const [path, setPath] = useState(initialPath); + const [draft, setDraft] = useState(DEFAULT_DRAFT); + const [active, setActive] = useState(null); + const [generation, setGeneration] = useState(0); + const log = useStreamLog(); + + const set = (key: K, value: DraftOptions[K]) => + setDraft((current) => ({ ...current, [key]: value })); + + const preview = useMemo(() => { + try { + return buildTiledWebSocketUrl( + connection.baseUrl, + resolveTiledPath(path, connection.initialPath), + { start: commit(draft).subscription.start }, + ); + } catch { + return '(invalid base URL)'; + } + }, [connection.baseUrl, connection.initialPath, path, draft]); + + const subscribe = () => { + log.pushNote('status', `subscribe ${path || '/'} with ${MODE_LABELS[mode]}`); + setActive(commit(draft)); + setGeneration((value) => value + 1); + }; + + const unsubscribe = () => { + log.pushNote('status', 'unsubscribed (component unmounted the subscription)'); + setActive(null); + }; + + const locked = active !== null; + const bodyProps: BodyProps = { + path, + connection, + options: active ?? commit(draft), + log, + socketFactory, + onSubscribeChild, + }; + + return ( +
+
+ + + {locked ? ( + + ) : ( + + )} + +
+ + + +

{preview}

+ + {active && ( +
+ {mode === 'container' && } + {mode === 'table' && } + {mode === 'array' && } + {mode === 'generic' && } + {mode === 'raw' && } +
+ )} + + +
+ ); +} + +function OptionsForm({ + mode, + draft, + set, + disabled, +}: { + mode: SubscriptionMode; + draft: DraftOptions; + set: (key: K, value: DraftOptions[K]) => void; + disabled: boolean; +}) { + const numberField = (key: keyof DraftOptions, label: string, placeholder: string) => ( + + ); + const checkbox = (key: keyof DraftOptions, label: string, title?: string) => ( + + ); + + return ( +
+ {numberField('start', 'start (replay from)', 'live only')} + {numberField('reconnectDelay', 'reconnect delay ms', '1000')} + {numberField('maxReconnectDelay', 'max delay ms', '30000')} + {numberField('maxReconnectAttempts', 'max attempts', 'unlimited')} + {checkbox('reconnect', 'auto-reconnect')} + {mode === 'table' && ( + <> + {checkbox( + 'accumulate', + 'accumulate rows', + 'Concatenate every payload into `rows` (re-renders per message).', + )} + {numberField('maxRows', 'maxRows', 'unbounded')} + + )} + {(mode === 'table' || mode === 'array') && + checkbox('trackLatest', 'trackLatest', 'Keep the newest message in state.')} + {mode === 'container' && + checkbox( + 'invalidateOnChange', + 'invalidate search cache on change', + 'Calls useTiledInvalidate().roots("search") from the callbacks, so the node browser refreshes.', + )} +
+ ); +} + +// #region bodies + +interface BodyProps { + path: string; + connection: TiledConnectionConfig; + options: CommittedOptions; + log: StreamLog; + socketFactory?: TiledSocketFactory; + onSubscribeChild: (path: string, mode: SubscriptionMode) => void; +} + +/** Callbacks every hook-based body passes, so lifecycle changes show up in the log. */ +function lifecycle(log: StreamLog) { + return { + onStatus: (status: TiledConnectionStatus) => log.pushNote('status', status), + onError: (error: TiledSubscriptionError) => + log.pushNote( + 'error', + `${error.kind}${error.code !== undefined ? ` (${error.code})` : ''}: ${error.message}`, + ), + }; +} + +interface ChildRow { + key: string; + structureFamily?: string; + specs: string[]; + createdAt?: string; + updatedAt?: string; + metadataKeys: string[]; + exitStatus?: string; +} + +function ContainerBody({ path, options, log, socketFactory, onSubscribeChild }: BodyProps) { + const invalidate = useTiledInvalidate(); + const [children, setChildren] = useState>({}); + + const upsert = ( + message: ContainerChildCreatedMessage | ContainerChildMetadataUpdatedMessage, + ) => { + setChildren((current) => { + const previous = current[message.key]; + const stop = message.metadata.stop as { exit_status?: string } | undefined; + const created = message.type === 'container-child-created'; + return { + ...current, + [message.key]: { + key: message.key, + structureFamily: created ? message.structure_family : previous?.structureFamily, + specs: message.specs.map((spec) => spec.name), + createdAt: created ? message.timestamp : previous?.createdAt, + updatedAt: created ? previous?.updatedAt : message.timestamp, + metadataKeys: Object.keys(message.metadata), + exitStatus: stop?.exit_status ?? previous?.exitStatus, + }, + }; + }); + if (options.invalidateOnChange) void invalidate.roots('search', 'metadata'); + }; + + const result = useTiledContainerSubscription(path, { + ...options.subscription, + ...lifecycle(log), + socketFactory, + onSchema: log.pushMessage, + onChildCreated(message) { + log.pushMessage(message); + upsert(message); + }, + onChildMetadataUpdated(message) { + log.pushMessage(message); + upsert(message); + }, + }); + + const rows = Object.values(children); + return ( + <> + +

+ Reports direct children only. Create a child, or update one's metadata, + and it appears here. Data written to a table or array further down does not; open + that node in its own panel. +

+ + + + + + + + + + + + + {rows.length === 0 && ( + + + + )} + {rows.map((row) => ( + + + + + + + + + + ))} + +
keyfamilyspecs + created (server) + + updated (server) + stop +
+ (no child events since subscribing) +
{row.key}{row.structureFamily ?? '?'}{row.specs.join(', ') || '-'}{row.createdAt?.slice(11, 19) ?? '-'}{row.updatedAt?.slice(11, 19) ?? '-'}{row.exitStatus ?? '-'} + +
+ + ); +} + +function TableBody({ path, options, log, socketFactory }: BodyProps) { + const result = useTiledTableSubscription(path, { + ...options.subscription, + ...lifecycle(log), + socketFactory, + accumulate: options.accumulate, + maxRows: options.maxRows, + trackLatest: options.trackLatest, + onSchema: log.pushMessage, + onData: log.pushMessage, + }); + + const rows = result.rows ?? (result.latest?.payload as Record | undefined); + const columns = rows ? Object.keys(rows) : []; + const length = columns.length ? (rows?.[columns[0]]?.length ?? 0) : 0; + const shownFrom = Math.max(0, length - 50); + + return ( + <> + +

+ {result.schema + ? `table-schema received (${result.schema.arrow_schema.length.toLocaleString()}-char Arrow schema; not decoded). It describes the columns, not existing rows: pass start=0 to replay retained rows.` + : 'waiting for table-schema…'} +

+

+ {options.accumulate + ? `accumulated rows: ${length.toLocaleString()}${options.maxRows ? ` (max ${options.maxRows})` : ''}` + : options.trackLatest + ? `latest payload: ${length} rows` + : 'callback mode: no state kept; see the log'} + {length > 50 && ` · showing the last 50`} +

+ {columns.length > 0 && ( +
+ + + + + {columns.map((column) => ( + + ))} + + + + {Array.from({ length: length - shownFrom }, (_, offset) => { + const index = shownFrom + offset; + return ( + + + {columns.map((column) => ( + + ))} + + ); + })} + +
# + {column} +
{index} + {formatCell(rows?.[column]?.[index])} +
+
+ )} + + ); +} + +function ArrayBody({ path, options, log, socketFactory }: BodyProps) { + const result = useTiledArraySubscription(path, { + ...options.subscription, + ...lifecycle(log), + socketFactory, + trackLatest: options.trackLatest, + onSchema: log.pushMessage, + onData: log.pushMessage, + onRef: log.pushMessage, + }); + const latest = result.latest; + const dtype = result.schema?.data_type; + + return ( + <> + +

+ {dtype + ? `dtype ${dtype.kind}${dtype.itemsize} (${dtype.endianness}). ` + : 'waiting for array-schema… '} + Every array-data message carries the written data in full; the server cannot leave + it out. Only data registered with put_data_source arrives as a lightweight + array-ref. +

+ {!options.trackLatest && ( +

+ trackLatest is off: nothing is kept in state. Messages only reach the log. +

+ )} + {latest?.type === 'array-data' && } + {latest?.type === 'array-ref' && } + + ); +} + +function ArrayDataView({ message }: { message: ArrayDataMessage }) { + return ( +
+ +
+
sequence
+
{message.sequence}
+
shape
+
[{message.shape.join(', ')}]
+
offset
+
{message.offset ? `[${message.offset.join(', ')}]` : 'null (full write)'}
+
block
+
{message.block ? `[${message.block.join(', ')}]` : 'null'}
+
mimetype
+
{message.mimetype}
+
+
+ ); +} + +function ArrayRefView({ message }: { message: ArrayRefMessage }) { + return ( +
+

+ array-ref #{message.sequence}: shape [{message.shape.join(', ')}] + {message.patch && ` · patch offset [${message.patch.offset.join(', ')}]`} +

+ + {message.uri} + +
+ ); +} + +function GenericBody({ path, options, log, socketFactory }: BodyProps) { + const result = useTiledSubscription(path, { + ...options.subscription, + ...lifecycle(log), + socketFactory, + onMessage: log.pushMessage, + }); + return ( + <> + +

+ lastMessage (re-renders once per message, because trackLastMessage defaults to + true): +

+ + + ); +} + +/** + * Drives `TiledSubscription` directly, as non-React code would. Unlike the hooks under a provider, + * this can send the connection bar's access token, so it is the way to test token auth. + */ +function RawBody({ path, connection, options, log, socketFactory }: BodyProps) { + const [subscription, setSubscription] = useState(null); + const [status, setStatus] = useState('idle'); + const [error, setError] = useState(); + const [, setTick] = useState(0); + + useEffect(() => { + const client = new TiledWebSocketClient({ + baseUrl: connection.baseUrl, + apiKey: connection.apiKey || null, + accessToken: connection.accessToken || null, + socketFactory, + }); + const next = client.subscribe( + resolveTiledPath(path, connection.initialPath), + options.subscription, + ); + const offs = [ + next.onStatus((value) => { + setStatus(value); + log.pushNote('status', value); + }), + next.onMessage((message) => { + log.pushMessage(message); + setTick((value) => value + 1); + }), + next.onError((value) => { + setError(value); + lifecycle(log).onError(value); + }), + ]; + setSubscription(next); + next.connect(); + return () => { + for (const off of offs) off(); + next.disconnect(); + }; + // The panel remounts this body on every Subscribe, so these are fixed for its lifetime. + // eslint-disable-next-line react-hooks/exhaustive-deps + }, []); + + return ( + <> + subscription?.disconnect()} + reconnect={() => { + setError(undefined); + subscription?.reconnect(); + }} + extra={ + <> + + + dropped {subscription?.getDroppedCount() ?? 0} ·{' '} + {connection.accessToken + ? 'auth frame: access_token' + : connection.apiKey + ? 'auth frame: api_key' + : 'no auth frame'} + + + } + /> + + ); +} + +// #endregion + +const STATUS_COLORS: Record = { + idle: 'bg-slate-300', + connecting: 'bg-amber-400', + reconnecting: 'bg-amber-500', + connected: 'bg-emerald-500', + closed: 'bg-slate-400', + error: 'bg-rose-500', +}; + +function StatusLine({ + status, + sequence, + log, + error, + disconnect, + reconnect, + extra, +}: { + status: TiledConnectionStatus; + /** What the hook returned as `sequence`. In callback mode it only refreshes on status changes. */ + sequence?: number; + log: StreamLog; + error?: TiledSubscriptionError; + disconnect: () => void; + reconnect: () => void; + extra?: React.ReactNode; +}) { + return ( +
+
+ + + {status} + + + last sequence delivered {log.lastSequence ?? '-'} + {log.lastSequence !== undefined && + ` · a reconnect asks for start=${log.lastSequence + 1}`} + {' · '} + + hook.sequence {sequence ?? '-'} + + + + + {extra} +
+ {error && ( +

+ {error.kind} + {error.code !== undefined ? ` (close ${error.code})` : ''}: {error.message} +

+ )} +
+ ); +} + +function formatCell(value: unknown): string { + if (value === null || value === undefined) return ''; + if (typeof value === 'number') + return Number.isInteger(value) ? String(value) : value.toPrecision(6); + return typeof value === 'object' ? JSON.stringify(value) : String(value); +} diff --git a/src/components/TiledWebsocket/TiledWebsocket.tsx b/src/components/TiledWebsocket/TiledWebsocket.tsx new file mode 100644 index 00000000..71b17163 --- /dev/null +++ b/src/components/TiledWebsocket/TiledWebsocket.tsx @@ -0,0 +1,241 @@ +import { useMemo, useState } from 'react'; +import { + TiledApiProvider, + createBrowserTokenStorage, + createMemoryTokenStorage, + createTiledApiClient, + useTiledMetadataQuery, + useTiledServerInfoQuery, +} from '@/api/tiled'; +import type { TiledSocketFactory } from '@/api/tiled'; +import { useTiledApiUrls } from '@/utils/apiUtils'; +import TiledConnectionBar, { + emptyTiledConnection, + type TiledConnectionConfig, +} from '../devtools/TiledConnectionBar'; +import TiledNodeBrowser from '../devtools/TiledNodeBrowser'; +import { + clearTiledConnection, + hasStoredTiledConnection, + loadTiledConnection, + saveTiledConnection, +} from '../devtools/tiledConnectionStorage'; +import { MODE_LABELS, modeForFamily, type SubscriptionMode } from './modes'; +import SubscriptionPanel from './SubscriptionPanel'; +import WriteHelpers from './WriteHelpers'; + +const CONNECTION_STORAGE = 'websocket'; + +export interface TiledWebsocketProps { + /** Overrides the browser `WebSocket` in every subscription; for tests. */ + socketFactory?: TiledSocketFactory; +} + +interface PanelEntry { + id: number; + path: string; + mode: SubscriptionMode; +} + +/** + * Manual test harness for Tiled's streaming websocket: the subscription hooks, the raw + * `TiledSubscription`, replay, reconnects, and the TanStack invalidation pattern. + * + * Like the query playground, it builds its **own client** from the connection bar and injects it + * through `TiledApiProvider`, so the hooks here resolve their URL, initial path and API key the same + * way an app's would. The one thing an injected client does not hand the hooks is an access token, + * so token auth is tested with the raw `TiledSubscription` mode, which reads the bar directly. + */ +export default function TiledWebsocket({ socketFactory }: TiledWebsocketProps) { + const configured = useTiledApiUrls(); + + const [applied, setApplied] = useState(() => + loadTiledConnection(CONNECTION_STORAGE, { + ...emptyTiledConnection(configured.httpBaseUrl), + apiKey: configured.apiKey ?? '', + }), + ); + + const client = useMemo(() => { + const next = createTiledApiClient({ + baseUrl: applied.baseUrl, + initialPath: applied.initialPath || undefined, + apiKey: applied.apiKey || null, + apiKeyScheme: applied.apiKeyScheme, + apiKeyLocation: applied.apiKeyLocation, + // Memory by default: the browser token keys are shared with the viewer. + tokenStorage: applied.useBrowserStorage + ? createBrowserTokenStorage() + : createMemoryTokenStorage(), + }); + if (applied.accessToken || applied.refreshToken) { + next.setSession({ + accessToken: applied.accessToken, + refreshToken: applied.refreshToken, + }); + } + return next; + }, [applied]); + + return ( + + { + setApplied(config); + saveTiledConnection(CONNECTION_STORAGE, config); + }} + socketFactory={socketFactory} + /> + + ); +} + +function HarnessBody({ + applied, + onApply, + socketFactory, +}: { + applied: TiledConnectionConfig; + onApply: (config: TiledConnectionConfig) => void; + socketFactory?: TiledSocketFactory; +}) { + const [stored, setStored] = useState(() => hasStoredTiledConnection(CONNECTION_STORAGE)); + const [path, setPath] = useState(''); + const [panels, setPanels] = useState([]); + const [nextId, setNextId] = useState(1); + + const info = useTiledServerInfoQuery({ staleTime: 30_000 }); + const node = useTiledMetadataQuery(path, { staleTime: 10_000 }); + const family = node.data?.attributes.structure_family; + const suggested = modeForFamily(path === '' ? 'container' : family); + + const addPanel = (panelPath: string, mode: SubscriptionMode) => { + setPanels((current) => [{ id: nextId, path: panelPath, mode }, ...current]); + setNextId((value) => value + 1); + }; + + return ( +
+ { + onApply(config); + setStored(true); + }} + onForget={ + stored + ? () => { + clearTiledConnection(CONNECTION_STORAGE); + setStored(false); + } + : undefined + } + status={ + <> + {info.isPending && 'connecting…'} + {info.data && + `Tiled ${info.data.library_version} · ${ + info.data.authentication?.required + ? 'auth required: the hooks send the API key as a first frame' + : 'anonymous reads allowed' + }`} + {info.data === null && !info.isPending && ( + + unreachable, or not a Tiled server + + )} + {' · '} + The key scheme and location only affect HTTP; sockets always send + credentials as a first frame. The server must have a streaming_cache, or + every subscription fails with a 500 (seen here as close 1006). + + } + /> + +
+
+
+

Node

+ +
+ + {path || '/'} is{' '} + {path === '' ? 'the root container' : (family ?? 'unknown')} + + +
+
+ + + +
+

Things to try

+
    +
  • + Subscribe to a container, then Create child container and{' '} + Add metadata.stop. Both events arrive, and the node browser + refreshes through cache invalidation. +
    • +
  • + Create an array, subscribe to it, then Write frame or{' '} + Append row. Note the offset on appended rows. +
    • +
  • + Press disconnect(), make a few writes, then{' '} + reconnect(). The missed events replay once, from{' '} + start = last + 1. +
    • +
  • + Subscribe with start = 0 to replay everything the streaming + cache still holds. +
    • +
  • + Restart Tiled while subscribed. The panel goes reconnecting, backs + off, then reconnects. With a memory cache, sequences restart at 1 + and the subscription starts a new epoch. +
    • +
  • + Use the raw TiledSubscription mode with a wrong API key on a + server that requires auth to see close 4003. +
    • +
+
+
+ +
+

+ Subscriptions{panels.length > 0 && ` (${panels.length})`} +

+ {panels.length === 0 && ( +

+ Pick a node and press Add subscription. Each panel holds its own + socket. +

+ )} + {panels.map((panel) => ( + + setPanels((current) => + current.filter((entry) => entry.id !== panel.id), + ) + } + /> + ))} +
+
+
+ ); +} diff --git a/src/components/TiledWebsocket/WriteHelpers.tsx b/src/components/TiledWebsocket/WriteHelpers.tsx new file mode 100644 index 00000000..8cdabe21 --- /dev/null +++ b/src/components/TiledWebsocket/WriteHelpers.tsx @@ -0,0 +1,336 @@ +import { useState } from 'react'; +import { useTiledApiClient } from '@/api/tiled'; +import type { PostMetadataRequest, PutMetadataRequest } from '@/api/tiled'; +import type { SubscriptionMode } from './modes'; + +interface CreatedNode { + path: string; + family: 'container' | 'array'; + /** Current `[rows, cols]` for arrays; it grows as rows are appended. */ + shape?: [number, number]; + stopped?: boolean; +} + +export interface WriteHelpersProps { + /** The container new nodes are created under. */ + parentPath: string; + onSubscribe: (path: string, mode: SubscriptionMode) => void; +} + +const buttonClass = + 'rounded border border-slate-300 px-2 py-1 text-xs hover:bg-slate-100 disabled:opacity-50'; + +/** + * Buttons that make the server emit stream events, so the harness can be driven without a Python + * session. They write over HTTP through the harness's own client, with the connection bar's key. + * + * Only containers and arrays are covered. A table append needs an Arrow IPC body, and Finch has no + * Arrow encoder (Tiled answers JSON with "No dispatch", and its CSV path is broken), so for tables + * the panel shows a Python snippet instead. + */ +export default function WriteHelpers({ parentPath, onSubscribe }: WriteHelpersProps) { + const client = useTiledApiClient(); + const [created, setCreated] = useState([]); + const [busy, setBusy] = useState(false); + const [result, setResult] = useState<{ ok: boolean; text: string } | null>(null); + const [blueskySpec, setBlueskySpec] = useState(true); + const [rows, setRows] = useState('32'); + const [cols, setCols] = useState('32'); + + const run = async (label: string, action: () => Promise) => { + setBusy(true); + try { + const detail = await action(); + setResult({ ok: true, text: detail ? `${label}: ${detail}` : `${label}: done` }); + } catch (error) { + setResult({ + ok: false, + text: `${label}: ${error instanceof Error ? error.message : String(error)}`, + }); + } finally { + setBusy(false); + } + }; + + const childPath = (key: string) => (parentPath ? `${parentPath}/${key}` : key); + const newKey = (prefix: string) => `${prefix}-${Date.now().toString(36)}`; + + const createContainer = () => + run('create container', async () => { + const key = newKey('ws-run'); + await client.createNode(parentPath, { + id: key, + structure_family: 'container', + metadata: { start: { uid: key, time: Date.now() / 1000 } }, + specs: blueskySpec ? [{ name: 'BlueskyRun', version: '3.0' }] : [], + data_sources: [], + } as unknown as PostMetadataRequest); + setCreated((current) => [...current, { path: childPath(key), family: 'container' }]); + return childPath(key); + }); + + const stopContainer = (node: CreatedNode) => + run('add metadata.stop', async () => { + const key = node.path.split('/').at(-1); + await client.updateMetadata(node.path, { + metadata: { + start: { uid: key }, + stop: { exit_status: 'success', time: Date.now() / 1000 }, + }, + } as unknown as PutMetadataRequest); + setCreated((current) => + current.map((entry) => (entry === node ? { ...entry, stopped: true } : entry)), + ); + }); + + const createArray = () => + run('create array', async () => { + const shape: [number, number] = [ + Math.max(1, Number(rows) || 32), + Math.max(1, Number(cols) || 32), + ]; + const key = newKey('ws-array'); + await client.createNode(parentPath, { + id: key, + structure_family: 'array', + metadata: {}, + specs: [], + data_sources: [ + { + structure_family: 'array', + structure: { + data_type: { + endianness: 'little', + kind: 'f', + itemsize: 8, + dt_units: null, + }, + chunks: [[shape[0]], [shape[1]]], + shape, + dims: null, + resizable: false, + }, + mimetype: 'application/x-zarr', + parameters: {}, + assets: [], + management: 'writable', + }, + ], + } as unknown as PostMetadataRequest); + setCreated((current) => [...current, { path: childPath(key), family: 'array', shape }]); + return `${childPath(key)} [${shape.join(' × ')}] float64`; + }); + + const writeFrame = (node: CreatedNode) => + run('write frame (PUT, full payload)', async () => { + const [height, width] = node.shape!; + await client.putArrayFull(node.path, testPattern(height, width)); + }); + + const appendRow = (node: CreatedNode) => + run('append row (PATCH, extend)', async () => { + const [height, width] = node.shape!; + const row = new Float64Array(width).map(() => Math.random()); + await client.patchArrayFull(node.path, row.buffer, { + offset: [height, 0], + shape: [1, width], + extend: true, + }); + setCreated((current) => + current.map((entry) => + entry === node ? { ...entry, shape: [height + 1, width] } : entry, + ), + ); + }); + + const deleteAll = () => + run('delete created nodes', async () => { + // Children before parents, in case one was created under another. + for (const node of [...created].reverse()) { + await client.deleteNode(node.path, { recursive: true, external_only: false }); + } + const count = created.length; + setCreated([]); + return `${count} deleted`; + }); + + return ( +
+

Make events happen

+

+ Writes under {parentPath || '/'} (the node + selected above) using the connection bar's API key. Subscribe to that container + first to see container-child-created. +

+ +
+ + +
+ +
+ + setRows(event.target.value)} + className="w-14 rounded border border-slate-300 px-1 py-0.5 font-mono text-xs" + /> + × + setCols(event.target.value)} + className="w-14 rounded border border-slate-300 px-1 py-0.5 font-mono text-xs" + /> +
+ + {created.length > 0 && ( +
    + {created.map((node) => ( +
  • + [{node.family}] + {node.path} + {node.shape && ( + [{node.shape.join(' × ')}] + )} + + {node.family === 'container' && ( + + )} + {node.family === 'array' && ( + <> + + + + )} +
    • + ))} +
  • + +
    • +
+ )} + + {result && ( +

+ {result.text} +

+ )} + + +
+ ); +} + +function TableSnippet({ parentPath }: { parentPath: string }) { + const [copied, setCopied] = useState(false); + const target = parentPath ? `c["${parentPath.split('/').join('"]["')}"]` : 'c'; + const snippet = `from tiled.client import from_uri +import pyarrow as pa, time + +c = from_uri("http://localhost:8000", api_key="…") +# SQL-backed, so it can be appended to (a write_table() Parquet table cannot). +t = ${target}.create_appendable_table( + pa.schema([("x", pa.int64()), ("y", pa.float64())]), key="ws-table" +) +for i in range(20): + # Pass a pyarrow Table: a pandas DataFrame drags its index in as an extra column. + t.append_partition(0, pa.table({"x": [i], "y": [i * 0.5]})) + time.sleep(0.5)`; + + return ( +
+ + Tables: append rows from Python + +

+ Subscribe to {parentPath ? `${parentPath}/ws-table` : 'ws-table'} with + useTiledTableSubscription once the table exists, then run the loop. +

+ 
+                {snippet}
+
+ +
+ ); +} + +/** A moving diagonal gradient, so successive frames look different. */ +function testPattern(height: number, width: number): ArrayBuffer { + const phase = Math.random() * Math.PI * 2; + const values = new Float64Array(height * width); + for (let y = 0; y < height; y += 1) { + for (let x = 0; x < width; x += 1) { + values[y * width + x] = Math.sin((x + y) / 4 + phase) + Math.random() * 0.2; + } + } + return values.buffer; +} diff --git a/src/components/TiledWebsocket/modes.ts b/src/components/TiledWebsocket/modes.ts new file mode 100644 index 00000000..56f2d509 --- /dev/null +++ b/src/components/TiledWebsocket/modes.ts @@ -0,0 +1,17 @@ +export type SubscriptionMode = 'container' | 'table' | 'array' | 'generic' | 'raw'; + +export const MODE_LABELS: Record = { + container: 'useTiledContainerSubscription', + table: 'useTiledTableSubscription', + array: 'useTiledArraySubscription', + generic: 'useTiledSubscription', + raw: 'TiledSubscription (no React)', +}; + +/** The subscription mode that fits a node's structure family. */ +export function modeForFamily(family: string | null | undefined): SubscriptionMode { + if (family === 'table') return 'table'; + if (family === 'array' || family === 'sparse') return 'array'; + if (family === 'container') return 'container'; + return 'generic'; +} diff --git a/src/components/TiledWebsocket/useStreamLog.ts b/src/components/TiledWebsocket/useStreamLog.ts new file mode 100644 index 00000000..056b9e88 --- /dev/null +++ b/src/components/TiledWebsocket/useStreamLog.ts @@ -0,0 +1,182 @@ +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import { isTiledSchemaMessage } from '@/api/tiled'; +import type { TiledStreamMessage } from '@/api/tiled'; + +/** One row in a subscription's event log. */ +export interface StreamLogEntry { + id: number; + receivedAt: number; + /** A message `type`, or `status` / `error` for lifecycle notes the harness adds itself. */ + kind: string; + sequence?: number; + summary: string; + /** The message as received, except that large array payloads are replaced by a placeholder. */ + detail?: unknown; +} + +export interface StreamLog { + entries: StreamLogEntry[]; + /** Messages received per type since the last clear. Lifecycle notes are not counted. */ + counts: Record; + total: number; + /** Sequence of the newest sequenced message delivered, whatever the hook mirrors into state. */ + lastSequence?: number; + /** Messages per second over the last few seconds. */ + rate: number; + pushMessage: (message: TiledStreamMessage) => void; + pushNote: (kind: 'status' | 'error', summary: string) => void; + clear: () => void; +} + +/** Above this many array elements, a payload is left out of the log entry. */ +const MAX_LOGGED_ELEMENTS = 2_000; +const RATE_WINDOW_MS = 5_000; + +/** + * A bounded, newest-first event log that re-renders at most once per animation frame. + * + * This is the pattern the hooks' docs recommend for high-frequency streams: messages land in a ref + * from a callback, and React state is updated on a schedule, not once per message. A burst of a + * hundred frames costs one render. + */ +export function useStreamLog(limit = 200): StreamLog { + const bufferRef = useRef([]); + const countsRef = useRef>({}); + const totalRef = useRef(0); + const lastSequenceRef = useRef(undefined); + const arrivalsRef = useRef([]); + const nextIdRef = useRef(1); + const frameRef = useRef(null); + + const [snapshot, setSnapshot] = useState({ + entries: [] as StreamLogEntry[], + counts: {} as Record, + total: 0, + lastSequence: undefined as number | undefined, + rate: 0, + }); + + const flush = useCallback(() => { + frameRef.current = null; + const now = Date.now(); + arrivalsRef.current = arrivalsRef.current.filter((at) => now - at < RATE_WINDOW_MS); + setSnapshot({ + entries: bufferRef.current.slice(), + counts: { ...countsRef.current }, + total: totalRef.current, + lastSequence: lastSequenceRef.current, + rate: arrivalsRef.current.length / (RATE_WINDOW_MS / 1000), + }); + }, []); + + const schedule = useCallback(() => { + if (frameRef.current !== null) return; + frameRef.current = requestAnimationFrame(flush); + }, [flush]); + + const add = useCallback( + (entry: Omit) => { + bufferRef.current.unshift({ + ...entry, + id: nextIdRef.current++, + receivedAt: Date.now(), + }); + if (bufferRef.current.length > limit) bufferRef.current.length = limit; + schedule(); + }, + [limit, schedule], + ); + + const pushMessage = useCallback( + (message: TiledStreamMessage) => { + countsRef.current[message.type] = (countsRef.current[message.type] ?? 0) + 1; + totalRef.current += 1; + arrivalsRef.current.push(Date.now()); + const sequence = isTiledSchemaMessage(message) ? undefined : message.sequence; + if (sequence !== undefined) lastSequenceRef.current = sequence; + add({ + kind: message.type, + sequence, + summary: summarizeMessage(message), + detail: withoutLargePayload(message), + }); + }, + [add], + ); + + const pushNote = useCallback( + (kind: 'status' | 'error', summary: string) => add({ kind, summary }), + [add], + ); + + const clear = useCallback(() => { + bufferRef.current = []; + countsRef.current = {}; + totalRef.current = 0; + lastSequenceRef.current = undefined; + arrivalsRef.current = []; + schedule(); + }, [schedule]); + + useEffect( + () => () => { + if (frameRef.current !== null) cancelAnimationFrame(frameRef.current); + }, + [], + ); + + return useMemo( + () => ({ ...snapshot, pushMessage, pushNote, clear }), + [snapshot, pushMessage, pushNote, clear], + ); +} + +/** One line describing a message, for the log. */ +export function summarizeMessage(message: TiledStreamMessage): string { + switch (message.type) { + case 'container-schema': + case 'ragged-schema': + return `schema v${message.version}`; + case 'table-schema': + return `schema v${message.version} · arrow schema ${message.arrow_schema.length.toLocaleString()} chars`; + case 'array-schema': { + const { kind, itemsize, endianness } = message.data_type; + return `dtype ${kind ?? '?'}${itemsize ?? ''} ${endianness ?? ''}`.trim(); + } + case 'container-child-created': { + const specs = message.specs.map((spec) => spec.name).join(', '); + return `${message.key} [${message.structure_family}]${specs ? ` · ${specs}` : ''}`; + } + case 'container-child-metadata-updated': { + const keys = Object.keys(message.metadata).join(', ') || '(empty)'; + const revision = + message.revision_number !== undefined ? ` · rev ${message.revision_number}` : ''; + return `${message.key} · metadata keys: ${keys}${revision}`; + } + case 'table-data': { + const columns = Object.keys(message.payload ?? {}); + const rows = columns.length ? (message.payload[columns[0]]?.length ?? 0) : 0; + return `${rows} row${rows === 1 ? '' : 's'} · ${columns.join(', ')} · partition ${ + message.partition ?? '-' + }${message.append ? ' · append' : ''}`; + } + case 'array-data': + return `shape [${message.shape.join(', ')}]${ + message.offset ? ` at offset [${message.offset.join(', ')}]` : ' · full write' + }`; + case 'array-ref': + return `ref shape [${message.shape.join(', ')}]${ + message.patch ? ` · patch at [${message.patch.offset.join(', ')}]` : '' + }`; + case 'ragged-data': + return 'ragged data'; + } +} + +function withoutLargePayload(message: TiledStreamMessage): unknown { + if (message.type !== 'array-data' && message.type !== 'ragged-data') return message; + const elements = + message.type === 'array-data' ? message.shape.reduce((product, n) => product * n, 1) : 0; + if (elements <= MAX_LOGGED_ELEMENTS) return message; + return { ...message, payload: `(${elements.toLocaleString()} values left out of the log)` }; +} diff --git a/src/components/devtools/tiledConnectionStorage.ts b/src/components/devtools/tiledConnectionStorage.ts index faad35a9..f8271983 100644 --- a/src/components/devtools/tiledConnectionStorage.ts +++ b/src/components/devtools/tiledConnectionStorage.ts @@ -36,9 +36,9 @@ const STORAGE_PREFIX = 'finch.devtools.tiledConnection.'; * * `playground` is `TiledQueryPlayground`. `endpoint-harness` is reserved for `TestTiled`, which * shares the bar but does not persist yet — the name is here so that wiring it up later cannot - * accidentally land on the playground's entry. + * accidentally land on the playground's entry. `websocket` is `TiledWebsocket`. */ -export type TiledConnectionStorageKey = 'playground' | 'endpoint-harness'; +export type TiledConnectionStorageKey = 'playground' | 'endpoint-harness' | 'websocket'; function storageKey(harness: TiledConnectionStorageKey): string { return `${STORAGE_PREFIX}${harness}`; diff --git a/src/index.ts b/src/index.ts index c7283662..ab15250c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -259,6 +259,15 @@ export { // graph useTiledGraphNodeEntitiesQuery, useTiledBindGraphEntityMutation, + // streaming (websocket) + useTiledSubscription, + useTiledContainerSubscription, + useTiledTableSubscription, + useTiledArraySubscription, + TiledWebSocketClient, + TiledSubscription, + buildTiledWebSocketUrl, + parseTiledStreamMessage, // client resolution, keys and invalidation useTiledClient, useTiledQueryScope, @@ -292,6 +301,18 @@ export type { FinchQueryOptions, FinchMutationOptions, TiledGraphOperationKind, + TiledStreamMessage, + TiledConnectionStatus, + TiledSubscriptionError, + TiledSubscriptionOptions, + TiledWebSocketClientOptions, + UseTiledSubscriptionOptions, + UseTiledSubscriptionResult, + UseTiledContainerSubscriptionOptions, + UseTiledTableSubscriptionOptions, + UseTiledTableSubscriptionResult, + UseTiledArraySubscriptionOptions, + UseTiledArraySubscriptionResult, } from './api/tiled'; // The cross-backend transport contract. Exported so that code implementing a Finch-shaped API layer // has something to conform to — see `src/api/shared/requestOptions.ts`. diff --git a/src/stories/TiledHooks.md b/src/stories/TiledHooks.md index 05234b7c..7562202f 100644 --- a/src/stories/TiledHooks.md +++ b/src/stories/TiledHooks.md @@ -179,6 +179,32 @@ awaited before `mutateAsync` resolves, so the affected queries have already refe --- +## Streams + +Websocket subscriptions to `/stream/single/{path}`. These are **not queries**: they hold a socket open +while mounted, take no part in the cache, and push messages to callbacks. To keep HTTP data fresh, +invalidate from a callback (`useTiledInvalidate().roots('search')`). Each one subscribes to **one +node**: a container reports only its direct children, so subscribe to a table or array directly to see +its data. Nothing is accumulated unless you opt in. + +| hook | arguments | delivers | +| ------------------------------- | ----------------------------------------------------------------- | ------------------------------------------------- | +| `useTiledSubscription` | `path, { onMessage?, trackLastMessage?, start?, enabled?, … }?` | every message; `status`, `lastMessage`, `sequence` | +| `useTiledContainerSubscription` | `path, { onChildCreated?, onChildMetadataUpdated?, onSchema? }?` | direct children created / metadata updated | +| `useTiledTableSubscription` | `path, { onData?, accumulate?, maxRows?, trackLatest? }?` | appended rows, columnar `{ column: values[] }` | +| `useTiledArraySubscription` | `path, { onData?, onRef?, trackLatest? }?` | written array data (full payload) or `array-ref` | + +```tsx +const invalidate = useTiledInvalidate(); +useTiledContainerSubscription('', { onChildCreated: () => invalidate.roots('search') }); +useTiledTableSubscription(`${uid}/primary/internal`, { onData: (m) => chart.append(m.payload) }); +``` + +See the Streaming section of `src/api/tiled/README.md` for auth, replay (`start`) and what the server +can and cannot leave out of a message. + +--- + ## Minimal examples ```tsx diff --git a/src/testing/tests/TiledSubscriptionHooks.test.tsx b/src/testing/tests/TiledSubscriptionHooks.test.tsx new file mode 100644 index 00000000..df821217 --- /dev/null +++ b/src/testing/tests/TiledSubscriptionHooks.test.tsx @@ -0,0 +1,302 @@ +import { act, renderHook } from '@testing-library/react'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { ReactNode } from 'react'; +import { + appendColumnarRows, + resetDefaultTiledApiClient, + useTiledArraySubscription, + useTiledContainerSubscription, + useTiledSubscription, + useTiledTableSubscription, +} from '../../api/tiled'; +import type { TiledClientLike } from '../../api/tiled'; +import { TiledApiProvider } from '../../api/tiled/runtime/TiledApiProvider'; +import { FinchConfigProvider } from '../../app/FinchConfigProvider'; +import { FakeSocket, STREAM_FIXTURES, fakeSocketFactory } from './tiledStreamFixtures'; + +const CONFIG = { tiledApiUrl: 'http://tiled.test/api/v1', tiledApiKey: 'cfg-key' }; + +function configWrapper(config: Record = CONFIG) { + return ({ children }: { children: ReactNode }) => ( + {children} + ); +} + +/** Open the latest socket and deliver `messages`, inside act so state updates flush. */ +function serve(...messages: object[]) { + act(() => { + const socket = FakeSocket.latest(); + if (socket.readyState === 0) socket.open(); + for (const message of messages) socket.emit(message); + }); +} + +beforeEach(() => { + FakeSocket.reset(); + resetDefaultTiledApiClient(); +}); + +afterEach(() => { + resetDefaultTiledApiClient(); + vi.restoreAllMocks(); +}); + +describe('useTiledSubscription', () => { + it('connects on mount with Finch config and closes on unmount', () => { + const { result, unmount } = renderHook( + () => useTiledSubscription('run/primary', { socketFactory: fakeSocketFactory }), + { wrapper: configWrapper() }, + ); + expect(FakeSocket.instances).toHaveLength(1); + const socket = FakeSocket.latest(); + expect(socket.url).toBe( + 'ws://tiled.test/api/v1/stream/single/run/primary?envelope_format=json', + ); + expect(result.current.status).toBe('connecting'); + + serve(STREAM_FIXTURES['table-schema']); + expect(JSON.parse(socket.sent[0])).toEqual({ type: 'auth', api_key: 'cfg-key' }); + expect(result.current.status).toBe('connected'); + expect(result.current.lastMessage?.type).toBe('table-schema'); + + unmount(); + expect(socket.closedWith).toBe(1000); + }); + + it('does not connect when disabled', () => { + const { result } = renderHook( + () => useTiledSubscription('x', { enabled: false, socketFactory: fakeSocketFactory }), + { wrapper: configWrapper() }, + ); + expect(FakeSocket.instances).toHaveLength(0); + expect(result.current.status).toBe('idle'); + }); + + it('opens a new socket when the path changes', () => { + const { rerender } = renderHook( + ({ path }) => useTiledSubscription(path, { socketFactory: fakeSocketFactory }), + { wrapper: configWrapper(), initialProps: { path: 'a' } }, + ); + const first = FakeSocket.latest(); + rerender({ path: 'b' }); + expect(FakeSocket.instances).toHaveLength(2); + expect(first.closedWith).toBe(1000); + expect(FakeSocket.latest().url).toContain('/stream/single/b?'); + }); + + it('does not reopen the socket when an inline callback changes', () => { + const { rerender } = renderHook( + () => + useTiledSubscription('a', { + socketFactory: (url) => fakeSocketFactory(url), + onMessage: () => {}, + }), + { wrapper: configWrapper() }, + ); + rerender(); + rerender(); + expect(FakeSocket.instances).toHaveLength(1); + }); + + it('calls the latest onMessage and tracks the sequence', () => { + const seen: string[] = []; + const { result } = renderHook( + () => + useTiledSubscription('t', { + socketFactory: fakeSocketFactory, + onMessage: (message) => seen.push(message.type), + }), + { wrapper: configWrapper() }, + ); + serve(STREAM_FIXTURES['table-schema'], STREAM_FIXTURES['table-data']); + expect(seen).toEqual(['table-schema', 'table-data']); + expect(result.current.sequence).toBe(2); + }); + + it('skips per-message state with trackLastMessage: false', () => { + const onMessage = vi.fn(); + const { result } = renderHook( + () => + useTiledSubscription('t', { + socketFactory: fakeSocketFactory, + trackLastMessage: false, + onMessage, + }), + { wrapper: configWrapper() }, + ); + serve(STREAM_FIXTURES['table-schema'], STREAM_FIXTURES['table-data']); + expect(onMessage).toHaveBeenCalledTimes(2); + expect(result.current.lastMessage).toBeUndefined(); + }); + + it('disconnect() closes, and reconnect() resumes from the last sequence', () => { + const { result } = renderHook( + () => useTiledSubscription('t', { socketFactory: fakeSocketFactory }), + { wrapper: configWrapper() }, + ); + serve(STREAM_FIXTURES['table-schema'], STREAM_FIXTURES['table-data']); + act(() => result.current.disconnect()); + expect(result.current.status).toBe('closed'); + + act(() => result.current.reconnect()); + expect(FakeSocket.instances).toHaveLength(2); + expect(new URL(FakeSocket.latest().url).searchParams.get('start')).toBe('3'); + }); + + it('applies an injected client URL, initial path and key', () => { + const injected = { + getBaseUrl: () => 'https://injected.test/api/v1', + getInitialPath: () => 'beamline', + getApiKey: () => 'inj-key', + } as unknown as TiledClientLike; + const wrapper = ({ children }: { children: ReactNode }) => ( + {children} + ); + renderHook(() => useTiledSubscription('run', { socketFactory: fakeSocketFactory }), { + wrapper, + }); + expect(FakeSocket.latest().url).toBe( + 'wss://injected.test/api/v1/stream/single/beamline/run?envelope_format=json', + ); + serve(); + expect(JSON.parse(FakeSocket.latest().sent[0])).toEqual({ + type: 'auth', + api_key: 'inj-key', + }); + }); + + it('stays idle for an injected stub with no URL', () => { + const injected = {} as TiledClientLike; + const wrapper = ({ children }: { children: ReactNode }) => ( + {children} + ); + const { result } = renderHook( + () => useTiledSubscription('run', { socketFactory: fakeSocketFactory }), + { wrapper }, + ); + expect(FakeSocket.instances).toHaveLength(0); + expect(result.current.status).toBe('idle'); + }); +}); + +describe('useTiledContainerSubscription', () => { + it('routes each container message to its callback', () => { + const onSchema = vi.fn(); + const onChildCreated = vi.fn(); + const onChildMetadataUpdated = vi.fn(); + renderHook( + () => + useTiledContainerSubscription('', { + socketFactory: fakeSocketFactory, + onSchema, + onChildCreated, + onChildMetadataUpdated, + }), + { wrapper: configWrapper() }, + ); + expect(FakeSocket.latest().url).toContain('/stream/single/?'); + serve( + STREAM_FIXTURES['container-schema'], + STREAM_FIXTURES['container-child-created'], + STREAM_FIXTURES['container-child-metadata-updated'], + ); + expect(onSchema).toHaveBeenCalledTimes(1); + expect(onChildCreated).toHaveBeenCalledWith( + expect.objectContaining({ key: 'ws-probe', sequence: 1 }), + ); + expect(onChildMetadataUpdated.mock.calls[0][0].metadata.stop).toEqual({ + exit_status: 'success', + }); + }); +}); + +describe('useTiledTableSubscription', () => { + const rowsAt = (sequence: number, x: number[]) => ({ + ...STREAM_FIXTURES['table-data'], + sequence, + payload: { x, y: x.map((v) => v * 10) }, + }); + + it('does not accumulate by default', () => { + const onData = vi.fn(); + const { result } = renderHook( + () => useTiledTableSubscription('t', { socketFactory: fakeSocketFactory, onData }), + { wrapper: configWrapper() }, + ); + serve(STREAM_FIXTURES['table-schema'], rowsAt(1, [1]), rowsAt(2, [2])); + expect(onData).toHaveBeenCalledTimes(2); + expect(result.current.schema?.type).toBe('table-schema'); + expect(result.current.rows).toBeUndefined(); + expect(result.current.latest).toBeUndefined(); + }); + + it('accumulates columnar rows when asked, trimmed to maxRows', () => { + const { result } = renderHook( + () => + useTiledTableSubscription('t', { + socketFactory: fakeSocketFactory, + accumulate: true, + maxRows: 3, + }), + { wrapper: configWrapper() }, + ); + expect(result.current.rows).toEqual({}); + serve(STREAM_FIXTURES['table-schema'], rowsAt(1, [1, 2]), rowsAt(2, [3, 4])); + expect(result.current.rows).toEqual({ x: [2, 3, 4], y: [20, 30, 40] }); + expect(result.current.latest?.sequence).toBe(2); + }); +}); + +describe('appendColumnarRows', () => { + it('pads columns that appear or disappear so lengths stay equal', () => { + const first = appendColumnarRows(undefined, { a: [1, 2] }); + expect(appendColumnarRows(first, { b: [9] })).toEqual({ + a: [1, 2, null], + b: [null, null, 9], + }); + }); +}); + +describe('useTiledArraySubscription', () => { + it('keeps no payload history by default', () => { + const onData = vi.fn(); + const onRef = vi.fn(); + const { result } = renderHook( + () => + useTiledArraySubscription('arr', { + socketFactory: fakeSocketFactory, + onData, + onRef, + }), + { wrapper: configWrapper() }, + ); + serve( + STREAM_FIXTURES['array-schema'], + STREAM_FIXTURES['array-ref'], + STREAM_FIXTURES['array-data'], + ); + expect(onRef).toHaveBeenCalledTimes(1); + expect(onData).toHaveBeenCalledWith(expect.objectContaining({ shape: [1, 3] })); + expect(result.current.schema?.data_type.kind).toBe('f'); + expect(result.current.latest).toBeUndefined(); + expect(result.current.shape).toBeUndefined(); + }); + + it('keeps only the newest message with trackLatest', () => { + const { result } = renderHook( + () => + useTiledArraySubscription('arr', { + socketFactory: fakeSocketFactory, + trackLatest: true, + }), + { wrapper: configWrapper() }, + ); + serve( + STREAM_FIXTURES['array-schema'], + STREAM_FIXTURES['array-ref'], + STREAM_FIXTURES['array-data'], + ); + expect(result.current.latest?.type).toBe('array-data'); + expect(result.current.shape).toEqual([1, 3]); + }); +}); diff --git a/src/testing/tests/TiledWebSocket.live.test.ts b/src/testing/tests/TiledWebSocket.live.test.ts new file mode 100644 index 00000000..a1928447 --- /dev/null +++ b/src/testing/tests/TiledWebSocket.live.test.ts @@ -0,0 +1,165 @@ +import axios from 'axios'; +import { createRequire } from 'node:module'; +import { afterAll, describe, expect, it } from 'vitest'; +import { TiledApiClient } from '../../api/tiled/client/TiledApiClient'; +import { TiledSubscription } from '../../api/tiled/client/websocket'; +import type { TiledStreamMessage, WebSocketLike } from '../../api/tiled/client/websocket'; + +// jsdom's WebSocket trips over its own Event class, so sockets come from the `ws` package (already +// installed transitively), whose browser-style `on*` properties fit WebSocketLike. +const NodeWebSocket = createRequire(import.meta.url)('ws') as new (url: string) => WebSocketLike; + +/** + * End-to-end against a real Tiled server. Skipped unless `TILED_LIVE_URL` is set: + * + * TILED_LIVE_URL=http://localhost:8000/api/v1 TILED_LIVE_API_KEY=secret \ + * npx vitest run TiledWebSocket.live + * + * The server needs a `streaming_cache`; without one every stream connection fails with a 500. Writes + * go under a throwaway key in `TILED_LIVE_PARENT` (default `finch-playground`), which is deleted + * afterwards. + * + * Sockets use the `ws` package and HTTP uses axios's Node adapter, so both reach the server directly, + * without jsdom's browser CORS checks in the way. + */ +const LIVE_URL = process.env.TILED_LIVE_URL; +const API_KEY = process.env.TILED_LIVE_API_KEY; +const PARENT = process.env.TILED_LIVE_PARENT ?? 'finch-playground'; +const KEY = `ws-live-${Date.now()}`; +const NODE = `${PARENT}/${KEY}`; + +const http = new TiledApiClient({ + baseUrl: LIVE_URL, + apiKey: API_KEY ?? null, + client: axios.create({ adapter: 'http' }), +}); + +/** Little-endian float64 bytes, matching the array's declared dtype. */ +const float64 = (values: number[]) => new Float64Array(values).buffer; + +/** Collect messages from a subscription and wait for one that matches. */ +function record(path: string, options: { start?: number } = {}) { + const messages: TiledStreamMessage[] = []; + const subscription = new TiledSubscription({ + baseUrl: LIVE_URL!, + path, + apiKey: API_KEY, + socketFactory: (url) => new NodeWebSocket(url), + ...options, + }); + subscription.onMessage((message) => messages.push(message)); + subscription.connect(); + + const waitFor = async ( + predicate: (message: TiledStreamMessage) => boolean, + timeoutMs = 5_000, + ): Promise => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const found = messages.find(predicate); + if (found) return found; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + throw new Error(`Timed out on ${path}; got ${JSON.stringify(messages.map((m) => m.type))}`); + }; + return { subscription, messages, waitFor }; +} + +describe.skipIf(!LIVE_URL)('Tiled websocket (live)', () => { + afterAll(async () => { + await http.deleteNode(NODE, { recursive: true, external_only: false }); + }); + + it('reports child creation and metadata updates on the parent', async () => { + const parent = record(PARENT); + await parent.waitFor((m) => m.type === 'container-schema'); + expect(parent.subscription.getStatus()).toBe('connected'); + + await http.createNode(PARENT, { + id: KEY, + structure_family: 'container', + metadata: { start: { uid: KEY } }, + specs: [], + data_sources: [], + } as never); + const created = await parent.waitFor( + (m) => m.type === 'container-child-created' && m.key === KEY, + ); + + await http.updateMetadata(NODE, { + metadata: { start: { uid: KEY }, stop: { exit_status: 'success' } }, + } as never); + const updated = await parent.waitFor( + (m) => m.type === 'container-child-metadata-updated' && m.key === KEY, + ); + expect( + updated.type === 'container-child-metadata-updated' && updated.metadata.stop, + ).toEqual({ + exit_status: 'success', + }); + if (created.type === 'container-child-created' && updated.type !== 'container-schema') { + expect(updated.sequence).toBeGreaterThan(created.sequence); + } + parent.subscription.disconnect(); + }); + + it('streams array writes, then replays them after a reconnect without duplicates', async () => { + const arrayPath = `${NODE}/arr`; + await http.createNode(NODE, { + id: 'arr', + structure_family: 'array', + metadata: {}, + specs: [], + data_sources: [ + { + structure_family: 'array', + structure: { + data_type: { endianness: 'little', kind: 'f', itemsize: 8, dt_units: null }, + chunks: [[2], [3]], + shape: [2, 3], + dims: null, + resizable: false, + }, + mimetype: 'application/x-zarr', + parameters: {}, + assets: [], + management: 'writable', + }, + ], + } as never); + + const live = record(arrayPath); + await live.waitFor((m) => m.type === 'array-schema'); + await http.putArrayFull(arrayPath, float64([1, 2, 3, 4, 5, 6])); + await http.putArrayFull(arrayPath, float64([7, 8, 9, 10, 11, 12])); + const first = await live.waitFor((m) => m.type === 'array-data'); + expect(first.type === 'array-data' && first.shape).toEqual([2, 3]); + await live.waitFor((m) => m.type === 'array-data' && m.sequence > first.sequence); + live.subscription.disconnect(); + + // A fresh subscription replaying from 0 sees both writes exactly once. + const replay = record(arrayPath, { start: 0 }); + await replay.waitFor((m) => m.type === 'array-data' && m.sequence > first.sequence); + const sequences = replay.messages + .filter((m) => m.type === 'array-data') + .map((m) => (m.type === 'array-data' ? m.sequence : -1)); + expect(new Set(sequences).size).toBe(sequences.length); + + // A forced reconnect resumes after the last sequence, so nothing repeats. + const before = replay.messages.length; + replay.subscription.reconnect(); + await replay.waitFor( + (m) => m.type === 'array-schema' && replay.messages.indexOf(m) >= before, + ); + await http.putArrayFull(arrayPath, float64([0, 0, 0, 0, 0, 0])); + await replay.waitFor( + (m) => m.type === 'array-data' && m.sequence === sequences.at(-1)! + 1, + ); + const after = replay.messages + .slice(before) + .filter((m) => m.type === 'array-data') + .map((m) => (m.type === 'array-data' ? m.sequence : -1)); + expect(after).toEqual([sequences.at(-1)! + 1]); + replay.subscription.disconnect(); + }); +}); diff --git a/src/testing/tests/TiledWebSocket.test.ts b/src/testing/tests/TiledWebSocket.test.ts new file mode 100644 index 00000000..6d2cc0c3 --- /dev/null +++ b/src/testing/tests/TiledWebSocket.test.ts @@ -0,0 +1,392 @@ +import { afterEach, beforeEach, describe, expect, expectTypeOf, it, vi } from 'vitest'; +import { + TILED_WS_CLOSE, + TiledSubscription, + TiledWebSocketClient, + buildTiledWebSocketUrl, + parseTiledStreamMessage, +} from '../../api/tiled/client/websocket'; +import type { + ContainerChildCreatedMessage, + TiledConnectionStatus, + TiledStreamMessage, + TiledSubscriptionConfig, + TiledSubscriptionError, +} from '../../api/tiled/client/websocket'; +import { FakeSocket, STREAM_FIXTURES, fakeSocketFactory } from './tiledStreamFixtures'; + +const API_URL = 'http://tiled.test:8000/api/v1'; + +function makeSubscription(overrides: Partial = {}) { + const subscription = new TiledSubscription({ + baseUrl: API_URL, + path: 'run/primary/internal', + socketFactory: fakeSocketFactory, + ...overrides, + }); + const statuses: TiledConnectionStatus[] = []; + const messages: TiledStreamMessage[] = []; + const errors: TiledSubscriptionError[] = []; + subscription.onStatus((status) => statuses.push(status)); + subscription.onMessage((message) => messages.push(message)); + subscription.onError((error) => errors.push(error)); + return { subscription, statuses, messages, errors }; +} + +/** Open the latest socket and deliver its schema, as the server does on every connection. */ +function connectLatest(schema: object = STREAM_FIXTURES['table-schema']): FakeSocket { + const socket = FakeSocket.latest(); + socket.open(); + socket.emit(schema); + return socket; +} + +beforeEach(() => { + FakeSocket.reset(); +}); + +afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +describe('buildTiledWebSocketUrl', () => { + it('swaps http for ws and https for wss', () => { + expect(buildTiledWebSocketUrl('http://localhost:8000/api/v1', 'a')).toBe( + 'ws://localhost:8000/api/v1/stream/single/a?envelope_format=json', + ); + expect(buildTiledWebSocketUrl('https://tiled.example.org/api/v1', 'a')).toBe( + 'wss://tiled.example.org/api/v1/stream/single/a?envelope_format=json', + ); + }); + + it('keeps the trailing slash the root container route needs', () => { + expect(buildTiledWebSocketUrl('http://localhost:8000/api/v1/', '')).toBe( + 'ws://localhost:8000/api/v1/stream/single/?envelope_format=json', + ); + expect(buildTiledWebSocketUrl(API_URL, '/')).toMatch(/\/stream\/single\/\?/); + }); + + it('encodes each segment but keeps separators', () => { + expect(buildTiledWebSocketUrl(API_URL, '/run 1//primary/a?b#c/')).toBe( + 'ws://tiled.test:8000/api/v1/stream/single/run%201/primary/a%3Fb%23c?envelope_format=json', + ); + }); + + it('adds envelope_format and start, including start=0', () => { + const url = new URL( + buildTiledWebSocketUrl(API_URL, 'x', { envelopeFormat: 'msgpack', start: 0 }), + ); + expect(url.searchParams.get('envelope_format')).toBe('msgpack'); + expect(url.searchParams.get('start')).toBe('0'); + }); + + it('never carries credentials', () => { + const client = new TiledWebSocketClient({ + baseUrl: API_URL, + apiKey: 'secret', + socketFactory: fakeSocketFactory, + }); + client.subscribe('x').connect(); + expect(FakeSocket.latest().url).not.toContain('secret'); + expect(FakeSocket.latest().url).not.toContain('api_key'); + }); +}); + +describe('parseTiledStreamMessage', () => { + it.each(Object.entries(STREAM_FIXTURES))('parses %s', (type, fixture) => { + const parsed = parseTiledStreamMessage(JSON.stringify(fixture)); + expect(parsed).toEqual(fixture); + expect(parsed?.type).toBe(type); + }); + + it.each([ + ['invalid JSON', '{nope'], + ['a non-object', '42'], + ['null', 'null'], + ['a missing type', '{"sequence": 1}'], + ['an unknown type', '{"type": "stream-closed", "sequence": 1}'], + ['a data message without a sequence', '{"type": "table-data", "payload": {}}'], + ])('drops %s', (_label, raw) => { + expect(parseTiledStreamMessage(raw)).toBeNull(); + }); + + it('drops binary frames', () => { + expect(parseTiledStreamMessage(new ArrayBuffer(4))).toBeNull(); + }); + + it('narrows on type without casts', () => { + const message = parseTiledStreamMessage( + JSON.stringify(STREAM_FIXTURES['container-child-created']), + ); + if (message?.type === 'container-child-created') { + expectTypeOf(message).toEqualTypeOf(); + expect(message.key).toBe('ws-probe'); + } else { + throw new Error('expected container-child-created'); + } + if (message.type === 'container-child-created') { + expectTypeOf(message.specs[0].name).toEqualTypeOf(); + } + }); +}); + +describe('TiledSubscription lifecycle', () => { + it('goes idle -> connecting -> connected and sends no auth frame without credentials', () => { + const { subscription, statuses } = makeSubscription(); + expect(subscription.getStatus()).toBe('idle'); + subscription.connect(); + expect(FakeSocket.instances).toHaveLength(1); + const socket = connectLatest(); + expect(statuses).toEqual(['idle', 'connecting', 'connected']); + expect(socket.sent).toEqual([]); + }); + + it('does not connect until connect() is called', () => { + new TiledWebSocketClient({ baseUrl: API_URL, socketFactory: fakeSocketFactory }).subscribe( + 'x', + ); + expect(FakeSocket.instances).toHaveLength(0); + }); + + it('sends a first-frame api key, preferring an access token when both are set', () => { + makeSubscription({ apiKey: 'k' }).subscription.connect(); + FakeSocket.latest().open(); + expect(JSON.parse(FakeSocket.latest().sent[0])).toEqual({ type: 'auth', api_key: 'k' }); + + makeSubscription({ apiKey: 'k', accessToken: 't' }).subscription.connect(); + FakeSocket.latest().open(); + expect(JSON.parse(FakeSocket.latest().sent[0])).toEqual({ + type: 'auth', + access_token: 't', + }); + }); + + it('treats 4003 as an auth error and does not reconnect', () => { + vi.useFakeTimers(); + const { subscription, errors } = makeSubscription({ apiKey: 'bad' }); + subscription.connect(); + FakeSocket.latest().open(); + FakeSocket.latest().serverClose(TILED_WS_CLOSE.AUTH_FAILED, 'Authentication failed'); + expect(subscription.getStatus()).toBe('error'); + expect(errors.at(-1)).toMatchObject({ kind: 'auth', code: 4003 }); + vi.advanceTimersByTime(60_000); + expect(FakeSocket.instances).toHaveLength(1); + }); + + it('treats a server close with 1000 as the end of the stream', () => { + vi.useFakeTimers(); + const { subscription } = makeSubscription(); + subscription.connect(); + connectLatest(); + FakeSocket.latest().serverClose(1000, 'Producer ended stream'); + expect(subscription.getStatus()).toBe('closed'); + vi.advanceTimersByTime(60_000); + expect(FakeSocket.instances).toHaveLength(1); + }); + + it('does not reconnect after an intentional disconnect', () => { + vi.useFakeTimers(); + const { subscription, statuses } = makeSubscription(); + subscription.connect(); + const socket = connectLatest(); + subscription.disconnect(); + expect(socket.closedWith).toBe(1000); + expect(statuses.at(-1)).toBe('closed'); + vi.advanceTimersByTime(60_000); + expect(FakeSocket.instances).toHaveLength(1); + }); + + it('reconnects after an unexpected drop with exponential backoff that resets on success', () => { + vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0.5); // no jitter + const { subscription, statuses } = makeSubscription({ + reconnectDelay: 100, + maxReconnectDelay: 350, + }); + subscription.connect(); + connectLatest(); + + FakeSocket.latest().serverClose(1006); + expect(subscription.getStatus()).toBe('reconnecting'); + vi.advanceTimersByTime(99); + expect(FakeSocket.instances).toHaveLength(1); + vi.advanceTimersByTime(1); + expect(FakeSocket.instances).toHaveLength(2); + + // The handshake fails twice: 200ms, then capped at 350ms. + FakeSocket.latest().serverClose(1006); + vi.advanceTimersByTime(200); + expect(FakeSocket.instances).toHaveLength(3); + FakeSocket.latest().serverClose(1006); + vi.advanceTimersByTime(349); + expect(FakeSocket.instances).toHaveLength(3); + vi.advanceTimersByTime(1); + expect(FakeSocket.instances).toHaveLength(4); + + // A successful connection resets the schedule. + connectLatest(); + expect(subscription.getStatus()).toBe('connected'); + FakeSocket.latest().serverClose(1006); + vi.advanceTimersByTime(100); + expect(FakeSocket.instances).toHaveLength(5); + expect(statuses).toContain('reconnecting'); + }); + + it('gives up after maxReconnectAttempts', () => { + vi.useFakeTimers(); + const { subscription, errors } = makeSubscription({ + reconnectDelay: 10, + maxReconnectAttempts: 2, + }); + subscription.connect(); + for (let i = 0; i < 3; i += 1) { + FakeSocket.latest().serverClose(1006); + vi.advanceTimersByTime(1_000); + } + expect(FakeSocket.instances).toHaveLength(3); + expect(subscription.getStatus()).toBe('error'); + expect(errors.at(-1)?.kind).toBe('reconnect-exhausted'); + }); + + it('does not reconnect when reconnect is false', () => { + vi.useFakeTimers(); + const { subscription } = makeSubscription({ reconnect: false }); + subscription.connect(); + FakeSocket.latest().serverClose(1006); + vi.advanceTimersByTime(60_000); + expect(FakeSocket.instances).toHaveLength(1); + expect(subscription.getStatus()).toBe('error'); + }); +}); + +describe('TiledSubscription sequences', () => { + const tableData = (sequence: number) => ({ ...STREAM_FIXTURES['table-data'], sequence }); + + it('tracks the last sequence and resumes at last + 1 on reconnect', () => { + vi.useFakeTimers(); + const { subscription } = makeSubscription({ start: 0 }); + subscription.connect(); + expect(new URL(FakeSocket.latest().url).searchParams.get('start')).toBe('0'); + connectLatest(); + FakeSocket.latest().emit(tableData(1)); + FakeSocket.latest().emit(tableData(2)); + expect(subscription.getLastSequence()).toBe(2); + + FakeSocket.latest().serverClose(1006); + vi.advanceTimersByTime(60_000); + expect(new URL(FakeSocket.latest().url).searchParams.get('start')).toBe('3'); + }); + + it('drops duplicate and older sequences within a connection', () => { + const { subscription, messages } = makeSubscription(); + subscription.connect(); + const socket = connectLatest(); + for (const sequence of [1, 2, 2, 1, 3]) socket.emit(tableData(sequence)); + expect(messages.filter((m) => m.type === 'table-data').map((m) => m.sequence)).toEqual([ + 1, 2, 3, + ]); + }); + + it('delivers the schema on every connection', () => { + vi.useFakeTimers(); + const { subscription, messages } = makeSubscription(); + subscription.connect(); + connectLatest(); + FakeSocket.latest().serverClose(1006); + vi.advanceTimersByTime(60_000); + connectLatest(); + expect(messages.filter((m) => m.type === 'table-schema')).toHaveLength(2); + }); + + it('starts a new epoch when the server counter resets between connections', () => { + vi.useFakeTimers(); + const { subscription, messages } = makeSubscription(); + subscription.connect(); + connectLatest(); + FakeSocket.latest().emit(tableData(7)); + FakeSocket.latest().serverClose(1006); + vi.advanceTimersByTime(60_000); + + // The server restarted with an in-memory cache, so its counter is back at 1. + connectLatest(); + FakeSocket.latest().emit(tableData(1)); + FakeSocket.latest().emit(tableData(1)); + FakeSocket.latest().emit(tableData(2)); + expect(messages.filter((m) => m.type === 'table-data').map((m) => m.sequence)).toEqual([ + 7, 1, 2, + ]); + expect(subscription.getLastSequence()).toBe(2); + }); + + it('forgets the sequence when the producer ends the stream', () => { + const { subscription } = makeSubscription(); + subscription.connect(); + connectLatest(); + FakeSocket.latest().emit(tableData(5)); + FakeSocket.latest().serverClose(1000); + subscription.connect(); + expect(new URL(FakeSocket.latest().url).searchParams.has('start')).toBe(false); + }); + + it('counts unparseable frames as dropped and keeps streaming', () => { + const { subscription, messages, errors } = makeSubscription(); + subscription.connect(); + const socket = connectLatest(); + socket.emit('{nope'); + socket.emitRaw(new ArrayBuffer(2)); + socket.emit(tableData(1)); + expect(subscription.getDroppedCount()).toBe(2); + expect(errors.filter((e) => e.kind === 'parse')).toHaveLength(2); + expect(messages.at(-1)?.type).toBe('table-data'); + }); +}); + +describe('TiledSubscription listeners', () => { + it('fans out to every listener, isolates a throwing one, and unsubscribes cleanly', () => { + vi.spyOn(console, 'error').mockImplementation(() => {}); + const subscription = new TiledSubscription({ + baseUrl: API_URL, + path: '', + socketFactory: fakeSocketFactory, + }); + const first = vi.fn(() => { + throw new Error('boom'); + }); + const second = vi.fn(); + const offFirst = subscription.onMessage(first); + subscription.onMessage(second); + subscription.connect(); + connectLatest(STREAM_FIXTURES['container-schema']); + expect(first).toHaveBeenCalledTimes(1); + expect(second).toHaveBeenCalledTimes(1); + + offFirst(); + FakeSocket.latest().emit(STREAM_FIXTURES['container-child-created']); + expect(first).toHaveBeenCalledTimes(1); + expect(second).toHaveBeenCalledTimes(2); + }); + + it('replays the current status to a new status listener', () => { + const subscription = new TiledSubscription({ + baseUrl: API_URL, + path: '', + socketFactory: fakeSocketFactory, + }); + subscription.connect(); + const listener = vi.fn(); + subscription.onStatus(listener); + expect(listener).toHaveBeenCalledWith('connecting'); + }); + + it('ignores events from a socket it has already replaced', () => { + const { subscription, messages } = makeSubscription(); + subscription.connect(); + const stale = connectLatest(); + subscription.reconnect(); + stale.emit({ ...STREAM_FIXTURES['table-data'], sequence: 9 }); + stale.serverClose(1006); + expect(messages.some((m) => m.type === 'table-data')).toBe(false); + expect(FakeSocket.instances).toHaveLength(2); + }); +}); diff --git a/src/testing/tests/TiledWebsocketHarness.test.tsx b/src/testing/tests/TiledWebsocketHarness.test.tsx new file mode 100644 index 00000000..0b0f4624 --- /dev/null +++ b/src/testing/tests/TiledWebsocketHarness.test.tsx @@ -0,0 +1,113 @@ +import { QueryClient, QueryClientProvider } from '@tanstack/react-query'; +import { act, fireEvent, render, screen, waitFor, within } from '@testing-library/react'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import TiledWebsocket from '../../components/TiledWebsocket/TiledWebsocket'; +import { FinchConfigProvider } from '../../app/FinchConfigProvider'; +import { FakeSocket, STREAM_FIXTURES, fakeSocketFactory } from './tiledStreamFixtures'; + +function renderHarness() { + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false, gcTime: 0 } }, + }); + return render( + + + + + , + ); +} + +/** Deliver frames, then let the log's animation-frame flush run. */ +async function serve(...messages: object[]) { + await act(async () => { + const socket = FakeSocket.latest(); + if (socket.readyState === 0) socket.open(); + for (const message of messages) socket.emit(message); + await new Promise((resolve) => requestAnimationFrame(() => resolve(undefined))); + }); +} + +beforeEach(() => { + FakeSocket.reset(); + window.localStorage.clear(); + // The node browser and server-info queries hit an unreachable host; their errors are expected. + vi.spyOn(console, 'error').mockImplementation(() => {}); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('TiledWebsocket harness', () => { + it('subscribes to a container and shows children as they are created and finished', async () => { + renderHarness(); + + fireEvent.click(screen.getByRole('button', { name: /Add subscription/ })); + const panel = screen.getByTestId('subscription-panel'); + expect(within(panel).getByLabelText('subscription mode')).toHaveValue('container'); + expect(FakeSocket.instances).toHaveLength(0); + + fireEvent.click(within(panel).getByRole('button', { name: 'Subscribe' })); + expect(FakeSocket.latest().url).toBe( + 'ws://tiled.test/api/v1/stream/single/?envelope_format=json', + ); + + await serve( + STREAM_FIXTURES['container-schema'], + STREAM_FIXTURES['container-child-created'], + ); + + expect(JSON.parse(FakeSocket.latest().sent[0])).toEqual({ type: 'auth', api_key: 'k' }); + expect(within(panel).getByTestId('subscription-status')).toHaveTextContent('connected'); + const row = within(panel).getByText('ws-probe').closest('tr')!; + expect(row).toHaveTextContent('BlueskyRun'); + + // An update carries the node's current specs and metadata, so the row takes them. + await serve(STREAM_FIXTURES['container-child-metadata-updated']); + expect(row).toHaveTextContent('success'); + expect(within(panel).getByTestId('stream-log')).toHaveTextContent( + 'container-child-metadata-updated', + ); + + // Drilling into a child opens a second panel on its path. + fireEvent.click(within(row).getByRole('button', { name: 'Subscribe' })); + const panels = screen.getAllByTestId('subscription-panel'); + expect(panels).toHaveLength(2); + expect(within(panels[0]).getByLabelText('subscription path')).toHaveValue('ws-probe'); + }); + + it('replays from the last sequence when reconnect() is pressed', async () => { + renderHarness(); + fireEvent.click(screen.getByRole('button', { name: /Add subscription/ })); + const panel = screen.getByTestId('subscription-panel'); + fireEvent.change(within(panel).getByLabelText('subscription mode'), { + target: { value: 'raw' }, + }); + fireEvent.click(within(panel).getByRole('button', { name: 'Subscribe' })); + await serve( + STREAM_FIXTURES['container-schema'], + STREAM_FIXTURES['container-child-created'], + ); + + fireEvent.click(within(panel).getByRole('button', { name: 'reconnect()' })); + expect(new URL(FakeSocket.latest().url).searchParams.get('start')).toBe('2'); + }); + + it('draws the newest array frame without keeping history', async () => { + renderHarness(); + fireEvent.click(screen.getByRole('button', { name: /Add subscription/ })); + const panel = screen.getByTestId('subscription-panel'); + fireEvent.change(within(panel).getByLabelText('subscription path'), { + target: { value: 'run/det' }, + }); + fireEvent.change(within(panel).getByLabelText('subscription mode'), { + target: { value: 'array' }, + }); + fireEvent.click(within(panel).getByRole('button', { name: 'Subscribe' })); + await serve(STREAM_FIXTURES['array-schema'], STREAM_FIXTURES['array-data']); + + await waitFor(() => expect(within(panel).getByTestId('array-preview')).toBeInTheDocument()); + expect(panel).toHaveTextContent('offset[2]'); + }); +}); diff --git a/src/testing/tests/tiledStreamFixtures.ts b/src/testing/tests/tiledStreamFixtures.ts new file mode 100644 index 00000000..dfa2a6c7 --- /dev/null +++ b/src/testing/tests/tiledStreamFixtures.ts @@ -0,0 +1,136 @@ +import type { TiledStreamMessage, WebSocketLike } from '../../api/tiled/client/websocket/types'; + +/** + * One envelope of every Tiled stream type, taken from frames a live Tiled 0.2.15 server sent with + * `envelope_format=json`. Payloads are trimmed but keep their shape. + */ +export const STREAM_FIXTURES = { + 'container-schema': { type: 'container-schema', version: 1 }, + 'container-child-created': { + type: 'container-child-created', + sequence: 1, + timestamp: '2026-10-06T18:56:21.633871', + key: 'ws-probe', + structure_family: 'container', + specs: [{ name: 'BlueskyRun', version: '3.0' }], + metadata: { start: { uid: 'abc' } }, + data_sources: [], + access_blob: {}, + }, + 'container-child-metadata-updated': { + type: 'container-child-metadata-updated', + key: 'ws-probe', + sequence: 2, + timestamp: '2026-10-06T18:56:21.657096', + specs: [], + metadata: { start: { uid: 'abc' }, stop: { exit_status: 'success' } }, + revision_number: 1, + }, + 'table-schema': { + type: 'table-schema', + version: 1, + arrow_schema: 'data:application/vnd.apache.arrow.file;base64,/////wADAAAQAAAA', + }, + 'table-data': { + type: 'table-data', + sequence: 2, + timestamp: '2026-10-06T18:56:37.326508', + mimetype: 'application/vnd.apache.arrow.file', + partition: 0, + append: true, + 'content-type': 'application/json', + payload: { x: [3], y: [2.5] }, + }, + 'array-schema': { + type: 'array-schema', + version: 1, + data_type: { endianness: 'little', kind: 'f', itemsize: 4, dt_units: null }, + }, + 'array-data': { + type: 'array-data', + sequence: 4, + timestamp: '2026-10-06T18:57:24.918589', + mimetype: 'application/octet-stream', + shape: [1, 3], + offset: [2], + block: null, + 'content-type': 'application/json', + payload: [[1.0, 1.0, 1.0]], + }, + 'array-ref': { + type: 'array-ref', + sequence: 3, + timestamp: '2026-10-06T18:57:30.000000', + data_source: { mimetype: 'application/x-hdf5', assets: [] }, + patch: { offset: [2, 0], shape: [1, 3] }, + shape: [3, 3], + uri: 'http://localhost:8000/api/v1/array/full/run/det?slice=2:3,0:3', + }, + 'ragged-schema': { type: 'ragged-schema', version: 1 }, + 'ragged-data': { + type: 'ragged-data', + sequence: 1, + timestamp: '2026-10-06T18:58:00.000000', + mimetype: 'application/json', + payload: [[1, 2], [3]], + }, +} satisfies { [K in TiledStreamMessage['type']]: Extract }; + +/** A scriptable stand-in for a browser `WebSocket`, matching the queue-server tests. */ +export class FakeSocket implements WebSocketLike { + static instances: FakeSocket[] = []; + + readyState = 0; + sent: string[] = []; + closedWith: number | undefined; + onopen: ((event: unknown) => void) | null = null; + onmessage: ((event: { data: unknown }) => void) | null = null; + onerror: ((event: unknown) => void) | null = null; + onclose: ((event: { code: number; reason?: string }) => void) | null = null; + + constructor(readonly url: string) { + FakeSocket.instances.push(this); + } + + static latest(): FakeSocket { + const socket = FakeSocket.instances.at(-1); + if (!socket) throw new Error('No socket has been opened.'); + return socket; + } + + static reset(): void { + FakeSocket.instances = []; + } + + send(data: string): void { + this.sent.push(data); + } + + close(code?: number): void { + this.readyState = 3; + this.closedWith = code; + } + + open(): void { + this.readyState = 1; + this.onopen?.({}); + } + + emit(payload: unknown): void { + this.onmessage?.({ + data: typeof payload === 'string' ? payload : JSON.stringify(payload), + }); + } + + emitRaw(data: unknown): void { + this.onmessage?.({ data }); + } + + serverClose(code: number, reason?: string): void { + this.readyState = 3; + this.onerror?.({}); + this.onclose?.({ code, reason }); + } +} + +export const fakeSocketFactory = (url: string): FakeSocket => new FakeSocket(url); diff --git a/src/vite-env.d.ts b/src/vite-env.d.ts index 2d1d62ac..4b722a84 100644 --- a/src/vite-env.d.ts +++ b/src/vite-env.d.ts @@ -4,8 +4,8 @@ interface ImportMetaEnv { readonly VITE_QSERVER_KEY: string; readonly VITE_QS_CONSOLE_URL: string; readonly VITE_PROXY_WS: string; - readonly VITE_API_TILED_URL: string; - readonly VITE_API_TILED_API_KEY: string; + readonly VITE_TILED_API_URL: string; + readonly VITE_TILED_API_KEY: string; // Add other environment variables here }