feat(subscription): add fastAssetCtxs subscription

src/api/subscription/_methods/fastAssetCtxs.ts

@@ -0,0 +1,117 @@
+import * as v from "@valibot/valibot";
+
+// ============================================================
+// API Schemas
+// ============================================================
+
+/**
+ * Subscription to mark and mid price events for all assets.
+ * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions
+ */
+export const FastAssetCtxsRequest = /* @__PURE__ */ (() => {
+  return v.object({
+    /** Type of subscription. */
+    type: v.literal("fastAssetCtxs"),
+  });
+})();
+export type FastAssetCtxsRequest = v.InferOutput<typeof FastAssetCtxsRequest>;
+
+/**
+ * Event of mark and mid prices, keyed by coin.
+ *
+ * The first message after subscribing is a full snapshot; later messages contain only the changed coins.
+ * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions
+ */
+export type FastAssetCtxsEvent = {
+  /** Mark and mid prices for a single asset. */
+  [coin: string]: {
+    /**
+     * Mark price.
+     * @pattern ^[0-9]+(\.[0-9]+)?$
+     */
+    markPx?: string;
+    /**
+     * Mid price.
+     * @pattern ^[0-9]+(\.[0-9]+)?$
+     */
+    midPx?: string | null;
+  };
+};
+
+// ============================================================
+// Execution Logic
+// ============================================================
+
+import { parse } from "../../../_base.ts";
+import type { ISubscription, TransportError } from "../../../transport/mod.ts";
+import type { SubscriptionConfig } from "./_base/mod.ts";
+
+/**
+ * Subscribe to mark and mid prices for all assets.
+ *
+ * NOTE: payloads are decompressed with [`DecompressionStream`](https://developer.mozilla.org/en-US/docs/Web/API/DecompressionStream),
+ * which React Native does not provide; add a [polyfill](https://www.npmjs.com/package/compression-streams-polyfill) to use this subscription there.
+ *
+ * @param config General configuration for Subscription API subscriptions.
+ * @param listener A callback function to be called when the event is received.
+ * @param onError An optional callback function to be called when the subscription fails.
+ * @return A request-promise that resolves with a {@link ISubscription} object to manage the subscription lifecycle.
+ *
+ * @throws {ValidationError} When the request parameters fail validation (before sending).
+ * @throws {TransportError} When the transport layer throws an error.
+ *
+ * @example
+ * ```ts
+ * import { WebSocketTransport } from "@nktkas/hyperliquid";
+ * import { fastAssetCtxs } from "@nktkas/hyperliquid/api/subscription";
+ *
+ * const transport = new WebSocketTransport();
+ *
+ * const sub = await fastAssetCtxs(
+ *   { transport },
+ *   (data) => console.log(data),
+ * );
+ * ```
+ *
+ * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions
+ */
+export function fastAssetCtxs(
+  config: SubscriptionConfig,
+  listener: (data: FastAssetCtxsEvent) => void,
+  onError?: (error: TransportError) => void,
+): Promise<ISubscription> {
+  const payload = parse(FastAssetCtxsRequest, { type: "fastAssetCtxs" });
+  // The server pushes each update as a base64 + raw DEFLATE (RFC 1951) compressed JSON string (assumed to be valid).
+  // Decompress sequentially so events reach the listener in arrival order.
+  let queue = Promise.resolve();
+  return config.transport.subscribe<string>(payload.type, payload, (e) => {
+    queue = queue.then(async () => listener(await decompress(e.detail)));
+  }, { onError });
+}
+
+/** Decode a base64 + raw DEFLATE (RFC 1951) payload into a {@linkcode FastAssetCtxsEvent}. */
+async function decompress(data: string): Promise<FastAssetCtxsEvent> {
+  const bytes = Uint8Array.from(atob(data), (c) => c.charCodeAt(0));
+
+  const stream = new DecompressionStream("deflate-raw");
+  const writer = stream.writable.getWriter();
+  // Do not await write/close before draining: backpressure on multi-chunk output would deadlock.
+  writer.write(bytes);
+  writer.close();
+
+  const reader = stream.readable.getReader();
+  const chunks: Uint8Array[] = [];
+  let result = await reader.read();
+  while (!result.done) {
+    chunks.push(result.value);
+    result = await reader.read();
+  }
+
+  const merged = new Uint8Array(chunks.reduce((total, chunk) => total + chunk.length, 0));
+  let offset = 0;
+  for (const chunk of chunks) {
+    merged.set(chunk, offset);
+    offset += chunk.length;
+  }
+  return JSON.parse(new TextDecoder().decode(merged));
+}

src/api/subscription/client.ts

@@ -36,6 +36,7 @@ import {
   type ClearinghouseStateEvent,
   type ClearinghouseStateParameters,
 } from "./_methods/clearinghouseState.ts";
+import { fastAssetCtxs, type FastAssetCtxsEvent } from "./_methods/fastAssetCtxs.ts";
 import { l2Book, type L2BookEvent, type L2BookParameters } from "./_methods/l2Book.ts";
 import { notification, type NotificationEvent, type NotificationParameters } from "./_methods/notification.ts";
 import { openOrders, type OpenOrdersEvent, type OpenOrdersParameters } from "./_methods/openOrders.ts";
@@ -461,6 +462,40 @@ export class SubscriptionClient<C extends SubscriptionConfig = SubscriptionConfi
     return clearinghouseState(this.config_, params, listener, onError);
   }
 
+  /**
+   * Subscribe to mark and mid prices for all assets.
+   *
+   * NOTE: payloads are decompressed with [`DecompressionStream`](https://developer.mozilla.org/en-US/docs/Web/API/DecompressionStream),
+   * which React Native does not provide; add a [polyfill](https://www.npmjs.com/package/compression-streams-polyfill) to use this subscription there.
+   *
+   * @param listener A callback function to be called when the event is received.
+   * @param onError An optional callback function to be called when the subscription fails.
+   * @return A request-promise that resolves with a {@link ISubscription} object to manage the subscription lifecycle.
+   *
+   * @throws {ValidationError} When the request parameters fail validation (before sending).
+   * @throws {TransportError} When the transport layer throws an error.
+   *
+   * @example
+   * ```ts
+   * import * as hl from "@nktkas/hyperliquid";
+   *
+   * const transport = new hl.WebSocketTransport();
+   * const client = new hl.SubscriptionClient({ transport });
+   *
+   * const sub = await client.fastAssetCtxs((data) => {
+   *   console.log(data);
+   * });
+   * ```
+   *
+   * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions
+   */
+  fastAssetCtxs(
+    listener: (data: FastAssetCtxsEvent) => void,
+    onError?: (error: TransportError) => void,
+  ): Promise<ISubscription> {
+    return fastAssetCtxs(this.config_, listener, onError);
+  }
+
   /**
    * Subscribe to L2 order book updates for a specific asset.
    *
@@ -1088,6 +1123,7 @@ export type {
   ClearinghouseStateEvent as ClearinghouseStateWsEvent,
   ClearinghouseStateParameters as ClearinghouseStateWsParameters,
 } from "./_methods/clearinghouseState.ts";
+export type { FastAssetCtxsEvent as FastAssetCtxsWsEvent } from "./_methods/fastAssetCtxs.ts";
 export type { L2BookEvent as L2BookWsEvent, L2BookParameters as L2BookWsParameters } from "./_methods/l2Book.ts";
 export type {
   NotificationEvent as NotificationWsEvent,

src/api/subscription/mod.ts

@@ -35,6 +35,7 @@ export * from "./_methods/assetCtxs.ts";
 export * from "./_methods/bbo.ts";
 export * from "./_methods/candle.ts";
 export * from "./_methods/clearinghouseState.ts";
+export * from "./_methods/fastAssetCtxs.ts";
 export * from "./_methods/l2Book.ts";
 export * from "./_methods/notification.ts";
 export * from "./_methods/openOrders.ts";

tests/api/subscription/fastAssetCtxs.test.ts

@@ -0,0 +1,19 @@
+import type { FastAssetCtxsEvent } from "@nktkas/hyperliquid/api/subscription";
+import { schemaCoverage } from "../_utils/schemaCoverage.ts";
+import { typeToJsonSchema } from "../_utils/typeToJsonSchema.ts";
+import { collectEventsOverTime, runTest } from "./_t.ts";
+
+const sourceFile = new URL("../../../src/api/subscription/_methods/fastAssetCtxs.ts", import.meta.url).pathname;
+const responseSchema = typeToJsonSchema(sourceFile, "FastAssetCtxsEvent");
+
+runTest({
+  name: "fastAssetCtxs",
+  mode: "api",
+  fn: async (_t, client) => {
+    const data = await collectEventsOverTime<FastAssetCtxsEvent>(async (cb) => {
+      await client.fastAssetCtxs(cb);
+    }, 10_000);
+
+    schemaCoverage(responseSchema, data);
+  },
+});