From 3aac3ab6e8e061e5d7a9a09109044cdb953376f3 Mon Sep 17 00:00:00 2001 From: Kyle Poole Date: Wed, 14 Oct 2026 09:51:59 -0700 Subject: [PATCH 1/6] feat(sdk): route processed Solana subscriptions to edge --- jest.config.js | 1 + package.json | 16 +- scripts/finalize-esm.cjs | 18 +++ src/index.ts | 1 + src/realtime.ts | 9 ++ src/scripts/buildSdk.ts | 3 +- src/sdk/Subscribe.ts | 59 ++++++++ src/sdk/edgeSubscription.ts | 74 ++++++++++ src/sdk/index.test.ts | 1 + src/sdk/index.ts | 27 ++-- src/sdk/realtime.integration.test.ts | 101 +++++++++++++ src/sdk/realtime.test.ts | 211 +++++++++++++++++++++++++++ src/sdk/realtime.ts | 55 +++++++ tsconfig.json | 7 +- 14 files changed, 567 insertions(+), 16 deletions(-) create mode 100644 scripts/finalize-esm.cjs create mode 100644 src/realtime.ts create mode 100644 src/sdk/edgeSubscription.ts create mode 100644 src/sdk/realtime.integration.test.ts create mode 100644 src/sdk/realtime.test.ts create mode 100644 src/sdk/realtime.ts diff --git a/jest.config.js b/jest.config.js index b7d1c5c..c1044c5 100644 --- a/jest.config.js +++ b/jest.config.js @@ -5,6 +5,7 @@ module.exports = merge.recursive( { verbose: false, testEnvironment: "node", + moduleNameMapper: { "^(\\.{1,2}/.*)\\.js$": "$1" }, modulePathIgnorePatterns: ["dist"], coverageProvider: "v8", collectCoverage: true, diff --git a/package.json b/package.json index 5fe273d..169ed03 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@codex-data/sdk", - "version": "2.3.79", + "version": "2.4.0", "engines": { "node": ">=17.5.0" }, @@ -13,7 +13,12 @@ "import": "./dist/index.mjs", "require": "./dist/index.js" }, - "./schema.graphql": "./schema.graphql" + "./schema.graphql": "./schema.graphql", + "./realtime": { + "types": "./dist/realtime.d.ts", + "import": "./dist/realtime.mjs", + "require": "./dist/realtime.js" + } }, "sideEffects": false, "scripts": { @@ -23,7 +28,7 @@ "build:types": "pnpm run build:cjs && pnpm run build:esm && pnpm run build:rename-esm", "build:cjs": "tsc", "build:esm": "tsc -p tsconfig.esm.json", - "build:rename-esm": "find dist-esm -name '*.js' -exec sh -c 'mv \"$1\" \"${1%.js}.mjs\"' _ {} \\; && cp -r dist-esm/* dist/ && rm -rf dist-esm", + "build:rename-esm": "find dist-esm -name '*.js' -exec sh -c 'mv \"$1\" \"${1%.js}.mjs\"' _ {} \\; && node scripts/finalize-esm.cjs && cp -r dist-esm/* dist/ && rm -rf dist-esm", "fetch:schema": "curl -s https://graph.codex.io/schema/latest.graphql --output src/resources/schema.graphql && cp src/resources/schema.graphql schema.graphql", "copy:schema": "cp src/resources/schema.graphql schema.graphql", "schema:diff": "tsx src/scripts/schemaDiff.ts", @@ -44,7 +49,10 @@ "dist/index.d.ts", "dist/sdk/", "schema.graphql", - "README.md" + "README.md", + "dist/realtime.js", + "dist/realtime.mjs", + "dist/realtime.d.ts" ], "keywords": [ "crypto", diff --git a/scripts/finalize-esm.cjs b/scripts/finalize-esm.cjs new file mode 100644 index 0000000..213fdb1 --- /dev/null +++ b/scripts/finalize-esm.cjs @@ -0,0 +1,18 @@ +const fs = require("node:fs"); +const path = require("node:path"); + +function rewrite(directory) { + for (const entry of fs.readdirSync(directory, { withFileTypes: true })) { + const filename = path.join(directory, entry.name); + if (entry.isDirectory()) rewrite(filename); + else if (entry.name.endsWith(".mjs")) { + const source = fs.readFileSync(filename, "utf8"); + fs.writeFileSync( + filename, + source.replace(/\bfrom (["'])(\.[^"']+)\.js\1/g, "from $1$2.mjs$1"), + ); + } + } +} + +rewrite("dist-esm"); diff --git a/src/index.ts b/src/index.ts index abb5672..a54b228 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,6 +1,7 @@ // Main SDK exports export { Codex } from "./sdk"; export type { ApiConfig, CleanupFunction } from "./sdk"; +export type { SubscriptionOptions } from "./sdk/realtime.js"; // Export GraphQL types for use in applications export * from "./sdk/generated/graphql"; diff --git a/src/realtime.ts b/src/realtime.ts new file mode 100644 index 0000000..d9bba94 --- /dev/null +++ b/src/realtime.ts @@ -0,0 +1,9 @@ +export { + createRealtimeClient, + DEFAULT_EDGE_REALTIME_URL, +} from "./sdk/realtime.js"; +export type { + RealtimeClient, + RealtimeClientOptions, + SubscriptionOptions, +} from "./sdk/realtime.js"; diff --git a/src/scripts/buildSdk.ts b/src/scripts/buildSdk.ts index 7a4313c..80c6215 100644 --- a/src/scripts/buildSdk.ts +++ b/src/scripts/buildSdk.ts @@ -43,7 +43,7 @@ class Subscription { this.subscriptionName }SubscriptionVariables, sink: Sink>) => this.sdk.subscribe(\`${graphqlString}\`, vars, sink);`; + }Subscription>>, options?: SubscriptionOptions) => this.sdk.subscribe(\`${graphqlString}\`, vars, sink, options);`; } } @@ -112,6 +112,7 @@ import { ${this.subscriptions.map((q) => " " + q.pre()).join(",\n")} } from "./generated/graphql"; import { Codex } from "./index"; +import type { SubscriptionOptions } from "./realtime.js"; import { ExecutionResult, Sink } from "graphql-ws"; export class Subscribe { diff --git a/src/sdk/Subscribe.ts b/src/sdk/Subscribe.ts index 2cc3932..50f65b4 100644 --- a/src/sdk/Subscribe.ts +++ b/src/sdk/Subscribe.ts @@ -62,12 +62,14 @@ import { OnUnconfirmedEventsCreatedSubscriptionVariables, } from "./generated/graphql"; import { Codex } from "./index"; +import type { SubscriptionOptions } from "./realtime.js"; export class Subscribe { constructor(private sdk: Codex) {} onBalanceUpdated = ( vars: OnBalanceUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnBalanceUpdated($walletAddress: String!) { @@ -77,10 +79,12 @@ export class Subscribe { }`, vars, sink, + options, ); onBarsUpdated = ( vars: OnBarsUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnBarsUpdated($commitmentLevel: [BarCommitmentLevel!], $pairId: String, $quoteToken: QuoteToken) { @@ -90,12 +94,14 @@ export class Subscribe { }`, vars, sink, + options, ); onDetailedPredictionEventStatsUpdated = ( vars: OnDetailedPredictionEventStatsUpdatedSubscriptionVariables, sink: Sink< ExecutionResult >, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnDetailedPredictionEventStatsUpdated($eventId: String!) { @@ -105,12 +111,14 @@ export class Subscribe { }`, vars, sink, + options, ); onDetailedPredictionMarketStatsUpdated = ( vars: OnDetailedPredictionMarketStatsUpdatedSubscriptionVariables, sink: Sink< ExecutionResult >, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnDetailedPredictionMarketStatsUpdated($marketId: String!) { @@ -120,10 +128,12 @@ export class Subscribe { }`, vars, sink, + options, ); onDetailedStatsUpdated = ( vars: OnDetailedStatsUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnDetailedStatsUpdated($pairId: String, $tokenOfInterest: TokenOfInterest) { @@ -133,10 +143,12 @@ export class Subscribe { }`, vars, sink, + options, ); onDetailedTokenStatsUpdated = ( vars: OnDetailedTokenStatsUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnDetailedTokenStatsUpdated($tokenId: String) { @@ -146,10 +158,12 @@ export class Subscribe { }`, vars, sink, + options, ); onEventLabelCreated = ( vars: OnEventLabelCreatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnEventLabelCreated($id: String) { @@ -177,10 +191,12 @@ export class Subscribe { `, vars, sink, + options, ); onEventsCreated = ( vars: OnEventsCreatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnEventsCreated($address: String, $commitmentLevel: [EventCommitmentLevel!], $id: String, $networkId: Int, $quoteToken: QuoteToken) { @@ -190,10 +206,12 @@ export class Subscribe { }`, vars, sink, + options, ); onEventsCreatedByMaker = ( vars: OnEventsCreatedByMakerSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnEventsCreatedByMaker($commitmentLevel: [EventCommitmentLevel!], $input: OnEventsCreatedByMakerInput!) { @@ -203,10 +221,12 @@ export class Subscribe { }`, vars, sink, + options, ); onFilterTokensUpdated = ( vars: OnFilterTokensUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnFilterTokensUpdated($excludeTokens: [String], $filters: TokenFilters, $limit: Int, $offset: Int, $phrase: String, $rankings: [TokenRanking], $statsType: TokenPairStatisticsType, $tokens: [String], $updatePeriod: Int, $useAggregatedStats: Boolean) { @@ -216,10 +236,12 @@ export class Subscribe { }`, vars, sink, + options, ); onHoldersUpdated = ( vars: OnHoldersUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnHoldersUpdated($tokenId: String!) { @@ -229,10 +251,12 @@ export class Subscribe { }`, vars, sink, + options, ); onLatestTokens = ( vars: OnLatestTokensSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnLatestTokens($id: String, $networkId: Int, $tokenAddress: String) { @@ -242,10 +266,12 @@ export class Subscribe { }`, vars, sink, + options, ); onLaunchpadTokenEvent = ( vars: OnLaunchpadTokenEventSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnLaunchpadTokenEvent($input: OnLaunchpadTokenEventInput) { @@ -255,10 +281,12 @@ export class Subscribe { }`, vars, sink, + options, ); onLaunchpadTokenEventBatch = ( vars: OnLaunchpadTokenEventBatchSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnLaunchpadTokenEventBatch($input: OnLaunchpadTokenEventBatchInput) { @@ -268,10 +296,12 @@ export class Subscribe { }`, vars, sink, + options, ); onNftAssetsCreated = ( vars: OnNftAssetsCreatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnNftAssetsCreated($address: String, $networkId: Int, $tokenId: String) { @@ -281,10 +311,12 @@ export class Subscribe { }`, vars, sink, + options, ); onNftEventsCreated = ( vars: OnNftEventsCreatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnNftEventsCreated($address: String, $networkId: Int) { @@ -294,10 +326,12 @@ export class Subscribe { }`, vars, sink, + options, ); onNftPoolEventsCreated = ( vars: OnNftPoolEventsCreatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnNftPoolEventsCreated($collectionAddress: String, $exchangeAddress: String, $networkId: Int, $poolAddress: String) { @@ -307,10 +341,12 @@ export class Subscribe { }`, vars, sink, + options, ); onPairMetadataUpdated = ( vars: OnPairMetadataUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnPairMetadataUpdated($id: String, $quoteToken: QuoteToken, $useNonLiquidityTokenAsQuoteToken: Boolean) { @@ -320,10 +356,12 @@ export class Subscribe { }`, vars, sink, + options, ); onPredictionEventBarsUpdated = ( vars: OnPredictionEventBarsUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnPredictionEventBarsUpdated($eventId: String!) { @@ -333,10 +371,12 @@ export class Subscribe { }`, vars, sink, + options, ); onPredictionMarketBarsUpdated = ( vars: OnPredictionMarketBarsUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnPredictionMarketBarsUpdated($marketId: String!) { @@ -346,10 +386,12 @@ export class Subscribe { }`, vars, sink, + options, ); onPredictionTradesCreated = ( vars: OnPredictionTradesCreatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnPredictionTradesCreated($input: OnPredictionTradesCreatedInput) { @@ -359,10 +401,12 @@ export class Subscribe { }`, vars, sink, + options, ); onPriceUpdated = ( vars: OnPriceUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnPriceUpdated($address: String, $networkId: Int, $sourcePairAddress: String, $useWeightedPrices: Boolean) { @@ -372,10 +416,12 @@ export class Subscribe { }`, vars, sink, + options, ); onPricesUpdated = ( vars: OnPricesUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnPricesUpdated($input: [OnPricesUpdatedInput!]!, $useWeightedPrices: Boolean) { @@ -385,10 +431,12 @@ export class Subscribe { }`, vars, sink, + options, ); onTokenBarsUpdated = ( vars: OnTokenBarsUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnTokenBarsUpdated($commitmentLevel: [BarCommitmentLevel!], $networkId: Int, $tokenId: String) { @@ -398,10 +446,12 @@ export class Subscribe { }`, vars, sink, + options, ); onTokenEventsCreated = ( vars: OnTokenEventsCreatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnTokenEventsCreated($commitmentLevel: [EventCommitmentLevel!], $input: OnTokenEventsCreatedInput!) { @@ -411,10 +461,12 @@ export class Subscribe { }`, vars, sink, + options, ); onTokenLifecycleEventsCreated = ( vars: OnTokenLifecycleEventsCreatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnTokenLifecycleEventsCreated($address: String, $networkId: Int) { @@ -424,10 +476,12 @@ export class Subscribe { }`, vars, sink, + options, ); onUnconfirmedBarsUpdated = ( vars: OnUnconfirmedBarsUpdatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnUnconfirmedBarsUpdated($pairId: String, $quoteToken: QuoteToken) { @@ -437,10 +491,12 @@ export class Subscribe { }`, vars, sink, + options, ); onUnconfirmedEventsCreated = ( vars: OnUnconfirmedEventsCreatedSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnUnconfirmedEventsCreated($address: String, $id: String, $quoteToken: QuoteToken) { @@ -450,10 +506,12 @@ export class Subscribe { }`, vars, sink, + options, ); onUnconfirmedEventsCreatedByMaker = ( vars: OnUnconfirmedEventsCreatedByMakerSubscriptionVariables, sink: Sink>, + options?: SubscriptionOptions, ) => this.sdk.subscribe( `subscription OnUnconfirmedEventsCreatedByMaker($input: OnUnconfirmedEventsCreatedByMakerInput!) { @@ -463,5 +521,6 @@ export class Subscribe { }`, vars, sink, + options, ); } diff --git a/src/sdk/edgeSubscription.ts b/src/sdk/edgeSubscription.ts new file mode 100644 index 0000000..b7834d0 --- /dev/null +++ b/src/sdk/edgeSubscription.ts @@ -0,0 +1,74 @@ +import { getOperationAST, Kind, parse, valueFromASTUntyped } from "graphql"; +import type { SubscribePayload } from "graphql-ws"; + +const SOLANA_NETWORK_ID = 1399811149; +const SUPPORTED_OPERATIONS = new Set([ + "onEventsCreated", + "onTokenEventsCreated", + "onBarsUpdated", + "onTokenBarsUpdated", +]); + +/** Validate the actual GraphQL arguments, including literals and variable defaults. */ +export function assertEdgeSubscription(payload: SubscribePayload): void { + function fail(): never { + throw new Error( + "useEdge requires a supported Solana subscription with only Processed commitmentLevel", + ); + } + const document = parse(payload.query); + const operation = getOperationAST(document, payload.operationName); + if (!operation || operation.operation !== "subscription") fail(); + // Edge executes a single root field; root fragments and conditional roots + // are not supported by its current customer transport. + const selections = operation.selectionSet.selections; + if (selections.length !== 1) fail(); + const field = selections[0]; + if (field.kind !== Kind.FIELD || !SUPPORTED_OPERATIONS.has(field.name.value)) + fail(); + if (field.directives?.length) fail(); + + const variables = { ...payload.variables }; + for (const definition of operation.variableDefinitions ?? []) { + const name = definition.variable.name.value; + if ( + variables[name] === undefined && + definition.defaultValue !== undefined + ) { + variables[name] = valueFromASTUntyped(definition.defaultValue); + } + } + const args: Record = Object.fromEntries( + (field.arguments ?? []).map((argument) => [ + argument.name.value, + valueFromASTUntyped(argument.value, variables), + ]), + ); + const levels = Array.isArray(args.commitmentLevel) + ? args.commitmentLevel + : [args.commitmentLevel]; + if (!levels.length || levels.some((level) => level !== "Processed")) fail(); + + const networks: unknown[] = []; + const collect = (value: unknown): void => { + if (!value || typeof value !== "object" || Array.isArray(value)) return; + for (const [name, item] of Object.entries(value)) { + if (name === "networkId" && item != null) networks.push(item); + if (["id", "pairId", "tokenId"].includes(name) && item != null) { + if (typeof item !== "string" || !/^[^:]+:\d+$/.test(item)) fail(); + networks.push(Number(item.slice(item.lastIndexOf(":") + 1))); + } + if (name === "input") collect(item); + } + }; + collect(args); + if ( + !networks.length || + networks.some((network) => network !== SOLANA_NETWORK_ID) + ) + fail(); + // Without an explicit quote, edge proxies this request back to Oregon. + if (field.name.value === "onBarsUpdated" && args.quoteToken == null) { + throw new Error("useEdge onBarsUpdated requires an explicit quoteToken"); + } +} diff --git a/src/sdk/index.test.ts b/src/sdk/index.test.ts index e8a664c..0f67196 100644 --- a/src/sdk/index.test.ts +++ b/src/sdk/index.test.ts @@ -448,6 +448,7 @@ describe("Codex", () => { variables, }, sink, + undefined, ); }); }); diff --git a/src/sdk/index.ts b/src/sdk/index.ts index 1ce1631..4ab078f 100644 --- a/src/sdk/index.ts +++ b/src/sdk/index.ts @@ -1,16 +1,16 @@ import { TypedDocumentNode } from "@graphql-typed-document-node/core"; import { GraphQLClient, Variables } from "graphql-request"; -import { - Client as GraphQLWsClient, - createClient, - ExecutionResult, - Sink, -} from "graphql-ws"; +import { ExecutionResult, Sink } from "graphql-ws"; import WebSocket from "isomorphic-ws"; import { invariant } from "./invariant"; import { Mutation } from "./Mutation"; import { Query } from "./Query"; +import { + createRealtimeClient, + RealtimeClient, + SubscriptionOptions, +} from "./realtime.js"; import { Subscribe } from "./Subscribe"; export type CleanupFunction = () => void; @@ -22,6 +22,9 @@ export type ApiConfig = { // wss://graph.codex.io/graphql apiRealtimeUrl: string; + // Optional deployment override for the edge transport. + apiEdgeRealtimeUrl?: string; + // true if you want to connect a websocket, sometimes useful to turn this off in certain environments ws?: boolean; @@ -46,7 +49,7 @@ const defaultConfig: ApiConfig = { export class Codex { private client: GraphQLClient; - private wsClient: GraphQLWsClient | undefined; + private wsClient: RealtimeClient | undefined; public queries: Query; public mutations: Mutation; public subscriptions: Subscribe; @@ -92,10 +95,11 @@ export class Codex { private createWebsocketClient(config: ApiConfig) { return config.ws - ? createClient({ + ? createRealtimeClient({ webSocketImpl: WebSocket, keepAlive: 10_000, // ping server every 10 seconds url: config.apiRealtimeUrl, + edgeUrl: config.apiEdgeRealtimeUrl, connectionParams: { Authorization: this.apiKey, ...config.wsHeaders, @@ -122,7 +126,8 @@ export class Codex { */ public updateConfig(incomingConfig: Partial) { // If we have an existing websocket connection, dispose of it. - if (this.wsClient) this.dispose(); + if (this.wsClient) void this.dispose(); + this.apiConfig = { ...this.apiConfig, ...incomingConfig }; // Merge the incoming config with the existing config. const config = this.parseConfig( @@ -151,7 +156,7 @@ export class Codex { * It's useful for when you need to modify some of the configuration, or when you're done with the class. */ public async dispose() { - this.wsClient?.dispose(); + await this.wsClient?.dispose(); } /** @@ -222,6 +227,7 @@ export class Codex { doc: string, args: TVars, sink: Sink>, + options?: SubscriptionOptions, ): CleanupFunction { if (!this.wsClient) { throw new Error("Websocket is not enabled in the config"); @@ -233,6 +239,7 @@ export class Codex { variables: args, }, sink, + options, ); return cleanup; } diff --git a/src/sdk/realtime.integration.test.ts b/src/sdk/realtime.integration.test.ts new file mode 100644 index 0000000..2ffbf00 --- /dev/null +++ b/src/sdk/realtime.integration.test.ts @@ -0,0 +1,101 @@ +import type { AddressInfo } from "node:net"; + +import WebSocket from "isomorphic-ws"; + +import { createRealtimeClient } from "./realtime"; + +it("delivers edge and normal results over separate authenticated sockets", async () => { + const requests: Array<{ source: string; type: string; payload: unknown }> = + []; + const servers = ["main", "edge"].map((source) => { + const server = new WebSocket.Server({ host: "127.0.0.1", port: 0 }); + server.on("connection", (socket) => { + socket.on("message", (data) => { + const message = JSON.parse(data.toString()); + requests.push({ source, type: message.type, payload: message.payload }); + if (message.type === "connection_init") { + socket.send(JSON.stringify({ type: "connection_ack" })); + } else if (message.type === "subscribe") { + socket.send( + JSON.stringify({ + id: message.id, + type: "next", + payload: { data: { source } }, + }), + ); + socket.send(JSON.stringify({ id: message.id, type: "complete" })); + } + }); + }); + return server; + }); + await Promise.all( + servers.map( + (server) => + new Promise((resolve, reject) => { + server.once("listening", resolve); + server.once("error", reject); + }), + ), + ); + const urls = servers.map( + (server) => `ws://127.0.0.1:${(server.address() as AddressInfo).port}`, + ); + const client = createRealtimeClient({ + url: urls[0], + edgeUrl: urls[1], + webSocketImpl: WebSocket, + connectionParams: { Authorization: "test-customer" }, + }); + const query = `subscription { onEventsCreated(id: "pair:1399811149", commitmentLevel: [Processed]) { events { transactionHash } } }`; + try { + expect(() => + client.subscribe( + { + query: `subscription { onPriceUpdated(networkId: 1) { priceUsd } }`, + }, + { next: () => {}, error: () => {}, complete: () => {} }, + { useEdge: true }, + ), + ).toThrow(); + expect(requests).toEqual([]); + const receive = (useEdge: boolean) => + new Promise((resolve, reject) => { + client.subscribe<{ source: string }>( + { query }, + { + next: (result) => resolve(result.data!.source), + error: reject, + complete: () => {}, + }, + { useEdge }, + ); + }); + await expect(Promise.all([receive(false), receive(true)])).resolves.toEqual( + ["main", "edge"], + ); + expect( + requests + .filter(({ type }) => type === "connection_init") + .map(({ payload }) => payload), + ).toEqual([ + { Authorization: "test-customer" }, + { Authorization: "test-customer" }, + ]); + expect( + requests + .filter(({ type }) => type === "subscribe") + .map(({ payload }) => payload), + ).toEqual([{ query }, { query }]); + } finally { + await client.dispose(); + await Promise.all( + servers.map( + (server) => + new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }), + ), + ); + } +}); diff --git a/src/sdk/realtime.test.ts b/src/sdk/realtime.test.ts new file mode 100644 index 0000000..555ba2b --- /dev/null +++ b/src/sdk/realtime.test.ts @@ -0,0 +1,211 @@ +import { type ClientOptions, createClient } from "graphql-ws"; + +import { EventCommitmentLevel } from "./generated/graphql"; +import { Codex } from "./index"; +import { createRealtimeClient, DEFAULT_EDGE_REALTIME_URL } from "./realtime"; + +jest.mock("graphql-ws", () => ({ createClient: jest.fn() })); + +const transports: Array<{ + options: ClientOptions; + subscribe: jest.Mock; + dispose: jest.Mock; +}> = []; +const sink = { next: jest.fn(), error: jest.fn(), complete: jest.fn() }; +const events = `subscription Events($input: OnTokenEventsCreatedInput!, $levels: [EventCommitmentLevel!]) { + onTokenEventsCreated(input: $input, commitmentLevel: $levels) { events { transactionHash } } +}`; +const variables = { + input: { networkId: 1399811149, tokenAddress: "token" }, + levels: ["Processed"], +}; + +beforeEach(() => { + transports.length = 0; + jest.mocked(createClient).mockImplementation((options) => { + const transport = { + options, + subscribe: jest.fn(() => jest.fn()), + dispose: jest.fn(async () => {}), + on: jest.fn(() => () => {}), + terminate: jest.fn(), + iterate: async function* () {}, + }; + transports.push(transport); + return transport; + }); +}); + +it("reuses one edge transport while keeping ordinary subscriptions on the main transport", () => { + const client = createRealtimeClient({ + url: "wss://main.example/graphql", + connectionParams: { Authorization: "customer" }, + }); + expect(transports).toHaveLength(1); + const payload = { query: events, variables }; + client.subscribe(payload, sink); + const unsubscribe = client.subscribe(payload, sink, { useEdge: true }); + client.subscribe(payload, sink, { useEdge: true }); + client.subscribe(payload, sink, { useEdge: false }); + expect(transports).toHaveLength(2); + expect(transports[0].subscribe).toHaveBeenCalledTimes(2); + expect(transports[1].subscribe).toHaveBeenCalledTimes(2); + expect(transports[1].options.url).toBe(DEFAULT_EDGE_REALTIME_URL); + expect(transports[1].options.connectionParams).toEqual({ + Authorization: "customer", + }); + expect(transports[1].subscribe).toHaveBeenCalledWith(payload, sink); + unsubscribe(); + expect(transports[1].subscribe.mock.results[0].value).toHaveBeenCalledTimes( + 1, + ); +}); + +it("uses current connection credentials and infrastructure options on both transports", async () => { + let token = "first"; + const connectionParams = async () => ({ Authorization: token }); + const client = createRealtimeClient({ + url: "wss://main.example/graphql", + edgeUrl: "wss://edge.staging.example/graphql", + connectionParams, + keepAlive: 10000, + lazyCloseTimeout: 5000, + }); + client.subscribe({ query: events, variables }, sink, { useEdge: true }); + token = "refreshed"; + for (const transport of transports) { + expect(transport.options.connectionParams).toBe(connectionParams); + await expect(connectionParams()).resolves.toEqual({ + Authorization: "refreshed", + }); + expect(transport.options.keepAlive).toBe(10000); + expect(transport.options.lazyCloseTimeout).toBe(5000); + } + expect(transports[1].options.url).toBe("wss://edge.staging.example/graphql"); +}); + +it.each([ + `subscription { onEventsCreated(id: "pair:1399811149", commitmentLevel: [Processed]) { events { transactionHash } } }`, + `subscription { onEventsCreated(address: "pair", networkId: 1399811149, commitmentLevel: Processed) { events { transactionHash } } }`, + `subscription { alias: onTokenEventsCreated(input: {networkId: 1399811149}, commitmentLevel: Processed) { events { transactionHash } } }`, + `subscription { onBarsUpdated(pairId: "pair:1399811149", quoteToken: token0, commitmentLevel: [Processed]) { timestamp } }`, + `subscription { onTokenBarsUpdated(tokenId: "token:1399811149", commitmentLevel: [Processed]) { timestamp } }`, + `subscription { onTokenBarsUpdated(networkId: 1399811149, commitmentLevel: [Processed]) { timestamp } }`, + `subscription($network: Int = 1399811149, $levels: [EventCommitmentLevel!] = [Processed]) { onEventsCreated(networkId: $network, commitmentLevel: $levels) { events { transactionHash } } }`, +])( + "accepts eligible literal and default-variable subscriptions: %s", + (query) => { + const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + client.subscribe({ query }, sink, { useEdge: true }); + expect(transports[1].subscribe).toHaveBeenCalledTimes(1); + }, +); + +it.each([ + {}, + { levels: undefined }, + { levels: null }, + { levels: [] }, + { levels: ["Confirmed"] }, + { levels: ["Preprocessed"] }, + { levels: ["Processed", "Confirmed"] }, + { levels: ["Processed", "Preprocessed"] }, + { input: { networkId: 8453 } }, + { input: { networkId: "1399811149" } }, + { input: { tokenAddress: "token" } }, +])( + "rejects ineligible variables without sending an operation: %j", + (override) => { + const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + // An empty override exercises missing variables rather than the valid base. + const values = Object.keys(override).length + ? { ...variables, ...override } + : {}; + expect(() => + client.subscribe({ query: events, variables: values }, sink, { + useEdge: true, + }), + ).toThrow("useEdge requires"); + expect(transports).toHaveLength(1); + expect(transports[0].subscribe).not.toHaveBeenCalled(); + }, +); + +it.each([ + `query { onEventsCreated(networkId: 1399811149, commitmentLevel: [Processed]) { events { transactionHash } } }`, + `subscription { onPriceUpdated(networkId: 1399811149, commitmentLevel: [Processed]) { priceUsd } }`, + `subscription { onEventsCreatedByMaker(input: {networkId: 1399811149}, commitmentLevel: [Processed]) { events { transactionHash } } }`, + `subscription { onEventsCreated(id: "pair:8453", commitmentLevel: [Processed]) { events { transactionHash } } }`, + `subscription { onEventsCreated(id: "pair:1399811149", networkId: 1, commitmentLevel: [Processed]) { events { transactionHash } } }`, + `subscription { onEventsCreated(commitmentLevel: [Processed]) { events { transactionHash } } }`, + `subscription { onTokenBarsUpdated(tokenId: "token", commitmentLevel: [Processed]) { timestamp } }`, + `subscription { a: onEventsCreated(networkId: 1399811149, commitmentLevel: [Processed]) { id } b: onEventsCreated(networkId: 1399811149, commitmentLevel: [Processed]) { id } }`, + `subscription { onEventsCreated(networkId: 1399811149, commitmentLevel: [Processed]) @skip(if: true) { id } }`, + `subscription { ...Root } fragment Root on Subscription { onEventsCreated(networkId: 1399811149, commitmentLevel: [Processed]) { id } }`, +])("rejects ineligible documents: %s", (query) => { + const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + expect(() => client.subscribe({ query }, sink, { useEdge: true })).toThrow( + "useEdge requires", + ); + expect(transports).toHaveLength(1); +}); + +it("rejects implicit pair quote selection instead of letting edge relay it through Oregon", () => { + const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + expect(() => + client.subscribe( + { + query: `subscription { onBarsUpdated(pairId: "pair:1399811149", commitmentLevel: [Processed]) { timestamp } }`, + }, + sink, + { useEdge: true }, + ), + ).toThrow("explicit quoteToken"); + expect(transports).toHaveLength(1); +}); + +it("validates the selected operation in a multi-operation document", () => { + const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + const query = `${events}\nsubscription Other { onPriceUpdated(networkId: 1) { priceUsd } }`; + client.subscribe({ query, variables, operationName: "Events" }, sink, { + useEdge: true, + }); + expect(() => + client.subscribe({ query, variables }, sink, { useEdge: true }), + ).toThrow(); +}); + +it("disposes both transports and prevents subscriptions from recreating them", async () => { + const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + client.subscribe({ query: events, variables }, sink, { useEdge: true }); + await client.dispose(); + expect(transports.map(({ dispose }) => dispose.mock.calls.length)).toEqual([ + 1, 1, + ]); + expect(() => + client.subscribe({ query: events, variables }, sink, { useEdge: true }), + ).toThrow("disposed"); +}); + +it("routes both raw and generated SDK subscriptions through useEdge", async () => { + const sdk = new Codex("customer"); + sdk.subscribe(events, variables, sink, { useEdge: true }); + sdk.subscriptions.onTokenEventsCreated( + { + input: variables.input, + commitmentLevel: [EventCommitmentLevel.Processed], + }, + sink, + { useEdge: true }, + ); + expect(transports).toHaveLength(2); + expect(transports[1].subscribe).toHaveBeenCalledTimes(2); + sdk.updateConfig({ wsHeaders: { Authorization: "new-customer" } }); + sdk.subscribe(events, variables, sink, { useEdge: true }); + expect(transports[3].options.connectionParams).toEqual({ + Authorization: "new-customer", + }); + expect(transports[1].dispose).toHaveBeenCalledTimes(1); + await sdk.dispose(); + expect(transports[3].dispose).toHaveBeenCalledTimes(1); +}); diff --git a/src/sdk/realtime.ts b/src/sdk/realtime.ts new file mode 100644 index 0000000..774e140 --- /dev/null +++ b/src/sdk/realtime.ts @@ -0,0 +1,55 @@ +import { + type Client, + type ClientOptions, + createClient, + type ExecutionResult, + type Sink, + type SubscribePayload, +} from "graphql-ws"; + +import { assertEdgeSubscription } from "./edgeSubscription.js"; + +export const DEFAULT_EDGE_REALTIME_URL = "wss://fra.edge.codex.io/graphql"; + +export interface SubscriptionOptions { + /** Experimental: deliver supported Processed Solana streams directly from edge. */ + useEdge?: boolean; +} + +export interface RealtimeClient extends Client { + subscribe, Extensions = unknown>( + payload: SubscribePayload, + sink: Sink>, + options?: SubscriptionOptions, + ): () => void; +} + +export interface RealtimeClientOptions extends ClientOptions { + /** Optional infrastructure override; callers select the transport with useEdge. */ + edgeUrl?: string; +} + +/** Lightweight transport shared by Codex and clients with their own GraphQL documents. */ +export function createRealtimeClient( + options: RealtimeClientOptions, +): RealtimeClient { + const { edgeUrl = DEFAULT_EDGE_REALTIME_URL, ...wsOptions } = options; + const main = createClient(wsOptions); + let edge: Client | undefined; + let disposed = false; + return { + ...main, + subscribe: (payload, sink, subscriptionOptions = {}) => { + if (disposed) throw new Error("Realtime client has been disposed"); + if (!subscriptionOptions.useEdge) return main.subscribe(payload, sink); + // Reject before creating an edge client or opening any upstream operation. + assertEdgeSubscription(payload); + edge ??= createClient({ ...wsOptions, url: edgeUrl, on: undefined }); + return edge.subscribe(payload, sink); + }, + dispose: async () => { + disposed = true; + await Promise.all([main.dispose(), edge?.dispose()]); + }, + }; +} diff --git a/tsconfig.json b/tsconfig.json index d205ffb..6101985 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -25,5 +25,10 @@ "**/*.spec.ts", "examples" ], - "include": ["src/sdk/**/*.ts", "src/index.ts", "!src/sdk/**/*.test.ts"] + "include": [ + "src/sdk/**/*.ts", + "src/index.ts", + "src/realtime.ts", + "!src/sdk/**/*.test.ts" + ] } From 3e32acd7a2fd02664b4d6aab8ab116a6c6d2e823 Mon Sep 17 00:00:00 2001 From: Kyle Poole Date: Wed, 14 Oct 2026 10:01:14 -0700 Subject: [PATCH 2/6] fix(sdk): follow existing import and build conventions --- jest.config.js | 1 - package.json | 2 +- scripts/finalize-esm.cjs | 18 ------------------ src/index.ts | 2 +- src/realtime.ts | 4 ++-- src/scripts/buildSdk.ts | 2 +- src/sdk/Subscribe.ts | 2 +- src/sdk/index.ts | 2 +- src/sdk/realtime.ts | 2 +- 9 files changed, 8 insertions(+), 27 deletions(-) delete mode 100644 scripts/finalize-esm.cjs diff --git a/jest.config.js b/jest.config.js index c1044c5..b7d1c5c 100644 --- a/jest.config.js +++ b/jest.config.js @@ -5,7 +5,6 @@ module.exports = merge.recursive( { verbose: false, testEnvironment: "node", - moduleNameMapper: { "^(\\.{1,2}/.*)\\.js$": "$1" }, modulePathIgnorePatterns: ["dist"], coverageProvider: "v8", collectCoverage: true, diff --git a/package.json b/package.json index 169ed03..22423c1 100644 --- a/package.json +++ b/package.json @@ -28,7 +28,7 @@ "build:types": "pnpm run build:cjs && pnpm run build:esm && pnpm run build:rename-esm", "build:cjs": "tsc", "build:esm": "tsc -p tsconfig.esm.json", - "build:rename-esm": "find dist-esm -name '*.js' -exec sh -c 'mv \"$1\" \"${1%.js}.mjs\"' _ {} \\; && node scripts/finalize-esm.cjs && cp -r dist-esm/* dist/ && rm -rf dist-esm", + "build:rename-esm": "find dist-esm -name '*.js' -exec sh -c 'mv \"$1\" \"${1%.js}.mjs\"' _ {} \\; && cp -r dist-esm/* dist/ && rm -rf dist-esm", "fetch:schema": "curl -s https://graph.codex.io/schema/latest.graphql --output src/resources/schema.graphql && cp src/resources/schema.graphql schema.graphql", "copy:schema": "cp src/resources/schema.graphql schema.graphql", "schema:diff": "tsx src/scripts/schemaDiff.ts", diff --git a/scripts/finalize-esm.cjs b/scripts/finalize-esm.cjs deleted file mode 100644 index 213fdb1..0000000 --- a/scripts/finalize-esm.cjs +++ /dev/null @@ -1,18 +0,0 @@ -const fs = require("node:fs"); -const path = require("node:path"); - -function rewrite(directory) { - for (const entry of fs.readdirSync(directory, { withFileTypes: true })) { - const filename = path.join(directory, entry.name); - if (entry.isDirectory()) rewrite(filename); - else if (entry.name.endsWith(".mjs")) { - const source = fs.readFileSync(filename, "utf8"); - fs.writeFileSync( - filename, - source.replace(/\bfrom (["'])(\.[^"']+)\.js\1/g, "from $1$2.mjs$1"), - ); - } - } -} - -rewrite("dist-esm"); diff --git a/src/index.ts b/src/index.ts index a54b228..f60173c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,7 +1,7 @@ // Main SDK exports export { Codex } from "./sdk"; export type { ApiConfig, CleanupFunction } from "./sdk"; -export type { SubscriptionOptions } from "./sdk/realtime.js"; +export type { SubscriptionOptions } from "./sdk/realtime"; // Export GraphQL types for use in applications export * from "./sdk/generated/graphql"; diff --git a/src/realtime.ts b/src/realtime.ts index d9bba94..cb03102 100644 --- a/src/realtime.ts +++ b/src/realtime.ts @@ -1,9 +1,9 @@ export { createRealtimeClient, DEFAULT_EDGE_REALTIME_URL, -} from "./sdk/realtime.js"; +} from "./sdk/realtime"; export type { RealtimeClient, RealtimeClientOptions, SubscriptionOptions, -} from "./sdk/realtime.js"; +} from "./sdk/realtime"; diff --git a/src/scripts/buildSdk.ts b/src/scripts/buildSdk.ts index 80c6215..f58b84f 100644 --- a/src/scripts/buildSdk.ts +++ b/src/scripts/buildSdk.ts @@ -112,7 +112,7 @@ import { ${this.subscriptions.map((q) => " " + q.pre()).join(",\n")} } from "./generated/graphql"; import { Codex } from "./index"; -import type { SubscriptionOptions } from "./realtime.js"; +import type { SubscriptionOptions } from "./realtime"; import { ExecutionResult, Sink } from "graphql-ws"; export class Subscribe { diff --git a/src/sdk/Subscribe.ts b/src/sdk/Subscribe.ts index 50f65b4..7d85aa8 100644 --- a/src/sdk/Subscribe.ts +++ b/src/sdk/Subscribe.ts @@ -62,7 +62,7 @@ import { OnUnconfirmedEventsCreatedSubscriptionVariables, } from "./generated/graphql"; import { Codex } from "./index"; -import type { SubscriptionOptions } from "./realtime.js"; +import type { SubscriptionOptions } from "./realtime"; export class Subscribe { constructor(private sdk: Codex) {} diff --git a/src/sdk/index.ts b/src/sdk/index.ts index 4ab078f..a2bb464 100644 --- a/src/sdk/index.ts +++ b/src/sdk/index.ts @@ -10,7 +10,7 @@ import { createRealtimeClient, RealtimeClient, SubscriptionOptions, -} from "./realtime.js"; +} from "./realtime"; import { Subscribe } from "./Subscribe"; export type CleanupFunction = () => void; diff --git a/src/sdk/realtime.ts b/src/sdk/realtime.ts index 774e140..c8a98b8 100644 --- a/src/sdk/realtime.ts +++ b/src/sdk/realtime.ts @@ -7,7 +7,7 @@ import { type SubscribePayload, } from "graphql-ws"; -import { assertEdgeSubscription } from "./edgeSubscription.js"; +import { assertEdgeSubscription } from "./edgeSubscription"; export const DEFAULT_EDGE_REALTIME_URL = "wss://fra.edge.codex.io/graphql"; From a9b33d9dfb9a9d4b0ea9a4866a0c260ede49e836 Mon Sep 17 00:00:00 2001 From: Kyle Poole Date: Wed, 14 Oct 2026 10:21:51 -0700 Subject: [PATCH 3/6] fix(sdk): preserve edge lifecycle and validated payloads --- src/sdk/edgeSubscription.ts | 3 +- src/sdk/realtime.integration.test.ts | 150 +++++++++++++++++++++++++++ src/sdk/realtime.test.ts | 30 ++++++ src/sdk/realtime.ts | 69 +++++++++++- 4 files changed, 247 insertions(+), 5 deletions(-) diff --git a/src/sdk/edgeSubscription.ts b/src/sdk/edgeSubscription.ts index b7834d0..30db689 100644 --- a/src/sdk/edgeSubscription.ts +++ b/src/sdk/edgeSubscription.ts @@ -47,7 +47,8 @@ export function assertEdgeSubscription(payload: SubscribePayload): void { const levels = Array.isArray(args.commitmentLevel) ? args.commitmentLevel : [args.commitmentLevel]; - if (!levels.length || levels.some((level) => level !== "Processed")) fail(); + if (!levels.length || [...levels].some((level) => level !== "Processed")) + fail(); const networks: unknown[] = []; const collect = (value: unknown): void => { diff --git a/src/sdk/realtime.integration.test.ts b/src/sdk/realtime.integration.test.ts index 2ffbf00..64b2660 100644 --- a/src/sdk/realtime.integration.test.ts +++ b/src/sdk/realtime.integration.test.ts @@ -1,5 +1,7 @@ +import { once } from "node:events"; import type { AddressInfo } from "node:net"; +import { parseMessage } from "graphql-ws"; import WebSocket from "isomorphic-ws"; import { createRealtimeClient } from "./realtime"; @@ -99,3 +101,151 @@ it("delivers edge and normal results over separate authenticated sockets", async ); } }); + +async function subscriptionServer() { + const server = new WebSocket.Server({ host: "127.0.0.1", port: 0 }); + server.on("connection", (socket) => { + socket.on("message", (data) => { + const message = parseMessage(data.toString()); + if (message.type === "connection_init") { + server.emit("init", message.payload); + socket.send(JSON.stringify({ type: "connection_ack" })); + } else if (message.type === "subscribe") { + server.emit("subscription", message.payload, socket); + } else if (message.type === "pong") { + server.emit("pong", message.payload); + } + }); + }); + await once(server, "listening"); + return { + server, + url: `ws://127.0.0.1:${(server.address() as AddressInfo).port}`, + close: () => + new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }), + }; +} + +const edgeQuery = `subscription Events($input: OnTokenEventsCreatedInput!, $levels: [EventCommitmentLevel!]) { + onTokenEventsCreated(input: $input, commitmentLevel: $levels) { events { transactionHash } } +}`; + +it("preserves the validated payload through delayed authentication and retries", async () => { + const edge = await subscriptionServer(); + let authenticate!: () => void; + const authentication = new Promise((resolve) => { + authenticate = resolve; + }); + let serializedLevels = 0; + const client = createRealtimeClient({ + url: "ws://main.example/graphql", + edgeUrl: edge.url, + webSocketImpl: WebSocket, + connectionParams: async () => { + await authentication; + return { Authorization: "customer", omit: "remove" }; + }, + jsonMessageReplacer: (key, value) => { + if (key === "levels") { + serializedLevels++; + return serializedLevels === 1 ? value : ["Confirmed"]; + } + return key === "omit" ? undefined : value; + }, + retryWait: async () => {}, + }); + const variables = { + input: { networkId: 1399811149, tokenAddress: "original" }, + levels: ["Processed"], + }; + const payload = { query: edgeQuery, variables }; + const expected = JSON.parse(JSON.stringify(payload)); + try { + const init = once(edge.server, "init"); + const first = once(edge.server, "subscription"); + client.subscribe( + payload, + { next: () => {}, error: () => {}, complete: () => {} }, + { useEdge: true }, + ); + variables.levels[0] = "Confirmed"; + variables.input.networkId = 1; + variables.input.tokenAddress = "changed"; + payload.query = "subscription { onPriceUpdated { priceUsd } }"; + authenticate(); + expect((await init)[0]).toEqual({ Authorization: "customer" }); + expect((await first)[0]).toEqual(expected); + const retry = once(edge.server, "subscription"); + client.terminate(); + expect((await retry)[0]).toEqual(expected); + expect(serializedLevels).toBe(1); + } finally { + authenticate(); + await client.dispose(); + await edge.close(); + } +}); + +it("forwards edge events, supports custom pong handling, and terminates both sockets", async () => { + const main = await subscriptionServer(); + const edge = await subscriptionServer(); + const configuredConnected = jest.fn(); + let activeSocket: WebSocket; + const client = createRealtimeClient({ + url: main.url, + edgeUrl: edge.url, + webSocketImpl: WebSocket, + disablePong: true, + retryWait: async () => {}, + on: { + connected: configuredConnected, + ping: (received) => { + if (received) activeSocket.send(JSON.stringify({ type: "pong" })); + }, + }, + }); + const connected = jest.fn(); + const mainConnected = jest.fn(); + const edgeConnected = jest.fn((socket) => { + activeSocket = socket as WebSocket; + }); + const closed = jest.fn(); + client.on("connected", connected); + client.on("connected", mainConnected, { useEdge: false }); + const removeEdge = client.on("connected", edgeConnected, { useEdge: true }); + client.on("closed", closed); + const payload = { + query: edgeQuery, + variables: { input: { networkId: 1399811149 }, levels: ["Processed"] }, + }; + const sink = { next: () => {}, error: () => {}, complete: () => {} }; + try { + const mainFirst = once(main.server, "subscription"); + const edgeFirst = once(edge.server, "subscription"); + client.subscribe(payload, sink); + client.subscribe(payload, sink, { useEdge: true }); + await mainFirst; + const [, edgeSocket] = await edgeFirst; + expect(configuredConnected).toHaveBeenCalledTimes(2); + expect(connected).toHaveBeenCalledTimes(2); + expect(mainConnected).toHaveBeenCalledTimes(1); + expect(edgeConnected).toHaveBeenCalledTimes(1); + const pong = once(edge.server, "pong"); + edgeSocket.send(JSON.stringify({ type: "ping" })); + await pong; + removeEdge(); + const mainRetry = once(main.server, "subscription"); + const edgeRetry = once(edge.server, "subscription"); + client.terminate(); + await Promise.all([mainRetry, edgeRetry]); + expect(closed).toHaveBeenCalledTimes(2); + expect(connected).toHaveBeenCalledTimes(4); + expect(mainConnected).toHaveBeenCalledTimes(2); + expect(edgeConnected).toHaveBeenCalledTimes(1); + } finally { + await client.dispose(); + await Promise.all([main.close(), edge.close()]); + } +}); diff --git a/src/sdk/realtime.test.ts b/src/sdk/realtime.test.ts index 555ba2b..df12082 100644 --- a/src/sdk/realtime.test.ts +++ b/src/sdk/realtime.test.ts @@ -1,5 +1,6 @@ import { type ClientOptions, createClient } from "graphql-ws"; +import { assertEdgeSubscription } from "./edgeSubscription"; import { EventCommitmentLevel } from "./generated/graphql"; import { Codex } from "./index"; import { createRealtimeClient, DEFAULT_EDGE_REALTIME_URL } from "./realtime"; @@ -106,6 +107,8 @@ it.each([ { levels: undefined }, { levels: null }, { levels: [] }, + { levels: new Array(1) }, + { levels: Object.assign(new Array(2), { 0: "Processed" }) }, { levels: ["Confirmed"] }, { levels: ["Preprocessed"] }, { levels: ["Processed", "Confirmed"] }, @@ -209,3 +212,30 @@ it("routes both raw and generated SDK subscriptions through useEdge", async () = await sdk.dispose(); expect(transports[3].dispose).toHaveBeenCalledTimes(1); }); + +it("rejects sparse commitment arrays even before wire serialization", () => { + expect(() => + assertEdgeSubscription({ + query: events, + variables: { ...variables, levels: new Array(1) }, + }), + ).toThrow("useEdge requires"); +}); + +it("validates serialized variables rather than a misleading toJSON value", () => { + const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + expect(() => + client.subscribe( + { + query: events, + variables: { + ...variables, + toJSON: () => ({ ...variables, levels: ["Confirmed"] }), + }, + }, + sink, + { useEdge: true }, + ), + ).toThrow("useEdge requires"); + expect(transports).toHaveLength(1); +}); diff --git a/src/sdk/realtime.ts b/src/sdk/realtime.ts index c8a98b8..1c124f7 100644 --- a/src/sdk/realtime.ts +++ b/src/sdk/realtime.ts @@ -2,6 +2,8 @@ import { type Client, type ClientOptions, createClient, + type Event, + type EventListener, type ExecutionResult, type Sink, type SubscribePayload, @@ -17,6 +19,12 @@ export interface SubscriptionOptions { } export interface RealtimeClient extends Client { + /** Observe both transports, or select one with useEdge. */ + on( + event: E, + listener: EventListener, + options?: SubscriptionOptions, + ): () => void; subscribe, Extensions = unknown>( payload: SubscribePayload, sink: Sink>, @@ -34,18 +42,71 @@ export function createRealtimeClient( options: RealtimeClientOptions, ): RealtimeClient { const { edgeUrl = DEFAULT_EDGE_REALTIME_URL, ...wsOptions } = options; - const main = createClient(wsOptions); + type EventHandlers = NonNullable; + const initialHandlers = wsOptions.on ? [wsOptions.on] : []; + const mainHandlers = new Set(initialHandlers); + const edgeHandlers = new Set(initialHandlers); + const handlersFor = (useEdge: boolean): EventHandlers => { + const handlers = () => Array.from(useEdge ? edgeHandlers : mainHandlers); + // Install before createClient so even non-lazy connection events are observed. + return { + connecting: (...args) => + handlers().forEach((on) => on?.connecting?.(...args)), + opened: (...args) => handlers().forEach((on) => on?.opened?.(...args)), + connected: (...args) => + handlers().forEach((on) => on?.connected?.(...args)), + ping: (...args) => handlers().forEach((on) => on?.ping?.(...args)), + pong: (...args) => handlers().forEach((on) => on?.pong?.(...args)), + message: (...args) => handlers().forEach((on) => on?.message?.(...args)), + closed: (...args) => handlers().forEach((on) => on?.closed?.(...args)), + error: (...args) => handlers().forEach((on) => on?.error?.(...args)), + }; + }; + const main = createClient({ ...wsOptions, on: handlersFor(false) }); let edge: Client | undefined; let disposed = false; return { ...main, + on: (event, listener, subscriptionOptions = {}) => { + const handlers = { [event]: listener }; + if (subscriptionOptions.useEdge !== true) mainHandlers.add(handlers); + if (subscriptionOptions.useEdge !== false) edgeHandlers.add(handlers); + return () => { + mainHandlers.delete(handlers); + edgeHandlers.delete(handlers); + }; + }, subscribe: (payload, sink, subscriptionOptions = {}) => { if (disposed) throw new Error("Realtime client has been disposed"); if (!subscriptionOptions.useEdge) return main.subscribe(payload, sink); // Reject before creating an edge client or opening any upstream operation. - assertEdgeSubscription(payload); - edge ??= createClient({ ...wsOptions, url: edgeUrl, on: undefined }); - return edge.subscribe(payload, sink); + const snapshot: SubscribePayload = JSON.parse( + JSON.stringify(payload, wsOptions.jsonMessageReplacer), + ); + assertEdgeSubscription(snapshot); + if (!edge) { + // Edge payloads are serialized once, before validation. Preserve custom + // serialization for handshakes without transforming subscriptions again. + let subscriptionMessage = false; + edge = createClient({ + ...wsOptions, + url: edgeUrl, + on: handlersFor(true), + jsonMessageReplacer: function (key, value) { + if (key === "") subscriptionMessage = value?.type === "subscribe"; + return subscriptionMessage + ? value + : wsOptions.jsonMessageReplacer + ? wsOptions.jsonMessageReplacer.call(this, key, value) + : value; + }, + }); + } + return edge.subscribe(snapshot, sink); + }, + terminate: () => { + main.terminate(); + edge?.terminate(); }, dispose: async () => { disposed = true; From af8b8e8f075aa7e309fcd41e3410ec387eb724ad Mon Sep 17 00:00:00 2001 From: Kyle Poole Date: Wed, 14 Oct 2026 10:31:59 -0700 Subject: [PATCH 4/6] refactor(sdk): name subscription routing client edge --- package.json | 14 ++++++------ src/edge.ts | 6 +++++ src/index.ts | 2 +- src/realtime.ts | 9 -------- src/scripts/buildSdk.ts | 2 +- src/sdk/Subscribe.ts | 2 +- ...ation.test.ts => edge.integration.test.ts} | 8 +++---- src/sdk/{realtime.test.ts => edge.test.ts} | 22 +++++++++---------- src/sdk/{realtime.ts => edge.ts} | 14 +++++------- src/sdk/index.ts | 14 +++++------- tsconfig.json | 2 +- 11 files changed, 43 insertions(+), 52 deletions(-) create mode 100644 src/edge.ts delete mode 100644 src/realtime.ts rename src/sdk/{realtime.integration.test.ts => edge.integration.test.ts} (97%) rename src/sdk/{realtime.test.ts => edge.test.ts} (92%) rename src/sdk/{realtime.ts => edge.ts} (90%) diff --git a/package.json b/package.json index 22423c1..9d58d51 100644 --- a/package.json +++ b/package.json @@ -14,10 +14,10 @@ "require": "./dist/index.js" }, "./schema.graphql": "./schema.graphql", - "./realtime": { - "types": "./dist/realtime.d.ts", - "import": "./dist/realtime.mjs", - "require": "./dist/realtime.js" + "./edge": { + "types": "./dist/edge.d.ts", + "import": "./dist/edge.mjs", + "require": "./dist/edge.js" } }, "sideEffects": false, @@ -50,9 +50,9 @@ "dist/sdk/", "schema.graphql", "README.md", - "dist/realtime.js", - "dist/realtime.mjs", - "dist/realtime.d.ts" + "dist/edge.js", + "dist/edge.mjs", + "dist/edge.d.ts" ], "keywords": [ "crypto", diff --git a/src/edge.ts b/src/edge.ts new file mode 100644 index 0000000..f468baf --- /dev/null +++ b/src/edge.ts @@ -0,0 +1,6 @@ +export { createEdgeClient, DEFAULT_EDGE_URL } from "./sdk/edge"; +export type { + EdgeClient, + EdgeClientOptions, + SubscriptionOptions, +} from "./sdk/edge"; diff --git a/src/index.ts b/src/index.ts index f60173c..c0dc84f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,7 +1,7 @@ // Main SDK exports export { Codex } from "./sdk"; export type { ApiConfig, CleanupFunction } from "./sdk"; -export type { SubscriptionOptions } from "./sdk/realtime"; +export type { SubscriptionOptions } from "./sdk/edge"; // Export GraphQL types for use in applications export * from "./sdk/generated/graphql"; diff --git a/src/realtime.ts b/src/realtime.ts deleted file mode 100644 index cb03102..0000000 --- a/src/realtime.ts +++ /dev/null @@ -1,9 +0,0 @@ -export { - createRealtimeClient, - DEFAULT_EDGE_REALTIME_URL, -} from "./sdk/realtime"; -export type { - RealtimeClient, - RealtimeClientOptions, - SubscriptionOptions, -} from "./sdk/realtime"; diff --git a/src/scripts/buildSdk.ts b/src/scripts/buildSdk.ts index f58b84f..c7bfaba 100644 --- a/src/scripts/buildSdk.ts +++ b/src/scripts/buildSdk.ts @@ -112,7 +112,7 @@ import { ${this.subscriptions.map((q) => " " + q.pre()).join(",\n")} } from "./generated/graphql"; import { Codex } from "./index"; -import type { SubscriptionOptions } from "./realtime"; +import type { SubscriptionOptions } from "./edge"; import { ExecutionResult, Sink } from "graphql-ws"; export class Subscribe { diff --git a/src/sdk/Subscribe.ts b/src/sdk/Subscribe.ts index 7d85aa8..b118f6f 100644 --- a/src/sdk/Subscribe.ts +++ b/src/sdk/Subscribe.ts @@ -1,6 +1,7 @@ // THIS FILE IS AUTOMATICALLY GENERATED, DO NOT EDIT IT DIRECTLY import { ExecutionResult, Sink } from "graphql-ws"; +import type { SubscriptionOptions } from "./edge"; import { OnBalanceUpdatedSubscription, OnBalanceUpdatedSubscriptionVariables, @@ -62,7 +63,6 @@ import { OnUnconfirmedEventsCreatedSubscriptionVariables, } from "./generated/graphql"; import { Codex } from "./index"; -import type { SubscriptionOptions } from "./realtime"; export class Subscribe { constructor(private sdk: Codex) {} diff --git a/src/sdk/realtime.integration.test.ts b/src/sdk/edge.integration.test.ts similarity index 97% rename from src/sdk/realtime.integration.test.ts rename to src/sdk/edge.integration.test.ts index 64b2660..6aa1a2a 100644 --- a/src/sdk/realtime.integration.test.ts +++ b/src/sdk/edge.integration.test.ts @@ -4,7 +4,7 @@ import type { AddressInfo } from "node:net"; import { parseMessage } from "graphql-ws"; import WebSocket from "isomorphic-ws"; -import { createRealtimeClient } from "./realtime"; +import { createEdgeClient } from "./edge"; it("delivers edge and normal results over separate authenticated sockets", async () => { const requests: Array<{ source: string; type: string; payload: unknown }> = @@ -43,7 +43,7 @@ it("delivers edge and normal results over separate authenticated sockets", async const urls = servers.map( (server) => `ws://127.0.0.1:${(server.address() as AddressInfo).port}`, ); - const client = createRealtimeClient({ + const client = createEdgeClient({ url: urls[0], edgeUrl: urls[1], webSocketImpl: WebSocket, @@ -139,7 +139,7 @@ it("preserves the validated payload through delayed authentication and retries", authenticate = resolve; }); let serializedLevels = 0; - const client = createRealtimeClient({ + const client = createEdgeClient({ url: "ws://main.example/graphql", edgeUrl: edge.url, webSocketImpl: WebSocket, @@ -193,7 +193,7 @@ it("forwards edge events, supports custom pong handling, and terminates both soc const edge = await subscriptionServer(); const configuredConnected = jest.fn(); let activeSocket: WebSocket; - const client = createRealtimeClient({ + const client = createEdgeClient({ url: main.url, edgeUrl: edge.url, webSocketImpl: WebSocket, diff --git a/src/sdk/realtime.test.ts b/src/sdk/edge.test.ts similarity index 92% rename from src/sdk/realtime.test.ts rename to src/sdk/edge.test.ts index df12082..d433d93 100644 --- a/src/sdk/realtime.test.ts +++ b/src/sdk/edge.test.ts @@ -1,9 +1,9 @@ import { type ClientOptions, createClient } from "graphql-ws"; +import { createEdgeClient, DEFAULT_EDGE_URL } from "./edge"; import { assertEdgeSubscription } from "./edgeSubscription"; import { EventCommitmentLevel } from "./generated/graphql"; import { Codex } from "./index"; -import { createRealtimeClient, DEFAULT_EDGE_REALTIME_URL } from "./realtime"; jest.mock("graphql-ws", () => ({ createClient: jest.fn() })); @@ -38,7 +38,7 @@ beforeEach(() => { }); it("reuses one edge transport while keeping ordinary subscriptions on the main transport", () => { - const client = createRealtimeClient({ + const client = createEdgeClient({ url: "wss://main.example/graphql", connectionParams: { Authorization: "customer" }, }); @@ -51,7 +51,7 @@ it("reuses one edge transport while keeping ordinary subscriptions on the main t expect(transports).toHaveLength(2); expect(transports[0].subscribe).toHaveBeenCalledTimes(2); expect(transports[1].subscribe).toHaveBeenCalledTimes(2); - expect(transports[1].options.url).toBe(DEFAULT_EDGE_REALTIME_URL); + expect(transports[1].options.url).toBe(DEFAULT_EDGE_URL); expect(transports[1].options.connectionParams).toEqual({ Authorization: "customer", }); @@ -65,7 +65,7 @@ it("reuses one edge transport while keeping ordinary subscriptions on the main t it("uses current connection credentials and infrastructure options on both transports", async () => { let token = "first"; const connectionParams = async () => ({ Authorization: token }); - const client = createRealtimeClient({ + const client = createEdgeClient({ url: "wss://main.example/graphql", edgeUrl: "wss://edge.staging.example/graphql", connectionParams, @@ -96,7 +96,7 @@ it.each([ ])( "accepts eligible literal and default-variable subscriptions: %s", (query) => { - const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + const client = createEdgeClient({ url: "wss://main.example/graphql" }); client.subscribe({ query }, sink, { useEdge: true }); expect(transports[1].subscribe).toHaveBeenCalledTimes(1); }, @@ -119,7 +119,7 @@ it.each([ ])( "rejects ineligible variables without sending an operation: %j", (override) => { - const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + const client = createEdgeClient({ url: "wss://main.example/graphql" }); // An empty override exercises missing variables rather than the valid base. const values = Object.keys(override).length ? { ...variables, ...override } @@ -146,7 +146,7 @@ it.each([ `subscription { onEventsCreated(networkId: 1399811149, commitmentLevel: [Processed]) @skip(if: true) { id } }`, `subscription { ...Root } fragment Root on Subscription { onEventsCreated(networkId: 1399811149, commitmentLevel: [Processed]) { id } }`, ])("rejects ineligible documents: %s", (query) => { - const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + const client = createEdgeClient({ url: "wss://main.example/graphql" }); expect(() => client.subscribe({ query }, sink, { useEdge: true })).toThrow( "useEdge requires", ); @@ -154,7 +154,7 @@ it.each([ }); it("rejects implicit pair quote selection instead of letting edge relay it through Oregon", () => { - const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + const client = createEdgeClient({ url: "wss://main.example/graphql" }); expect(() => client.subscribe( { @@ -168,7 +168,7 @@ it("rejects implicit pair quote selection instead of letting edge relay it throu }); it("validates the selected operation in a multi-operation document", () => { - const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + const client = createEdgeClient({ url: "wss://main.example/graphql" }); const query = `${events}\nsubscription Other { onPriceUpdated(networkId: 1) { priceUsd } }`; client.subscribe({ query, variables, operationName: "Events" }, sink, { useEdge: true, @@ -179,7 +179,7 @@ it("validates the selected operation in a multi-operation document", () => { }); it("disposes both transports and prevents subscriptions from recreating them", async () => { - const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + const client = createEdgeClient({ url: "wss://main.example/graphql" }); client.subscribe({ query: events, variables }, sink, { useEdge: true }); await client.dispose(); expect(transports.map(({ dispose }) => dispose.mock.calls.length)).toEqual([ @@ -223,7 +223,7 @@ it("rejects sparse commitment arrays even before wire serialization", () => { }); it("validates serialized variables rather than a misleading toJSON value", () => { - const client = createRealtimeClient({ url: "wss://main.example/graphql" }); + const client = createEdgeClient({ url: "wss://main.example/graphql" }); expect(() => client.subscribe( { diff --git a/src/sdk/realtime.ts b/src/sdk/edge.ts similarity index 90% rename from src/sdk/realtime.ts rename to src/sdk/edge.ts index 1c124f7..be0247c 100644 --- a/src/sdk/realtime.ts +++ b/src/sdk/edge.ts @@ -11,14 +11,14 @@ import { import { assertEdgeSubscription } from "./edgeSubscription"; -export const DEFAULT_EDGE_REALTIME_URL = "wss://fra.edge.codex.io/graphql"; +export const DEFAULT_EDGE_URL = "wss://fra.edge.codex.io/graphql"; export interface SubscriptionOptions { /** Experimental: deliver supported Processed Solana streams directly from edge. */ useEdge?: boolean; } -export interface RealtimeClient extends Client { +export interface EdgeClient extends Client { /** Observe both transports, or select one with useEdge. */ on( event: E, @@ -32,16 +32,14 @@ export interface RealtimeClient extends Client { ): () => void; } -export interface RealtimeClientOptions extends ClientOptions { +export interface EdgeClientOptions extends ClientOptions { /** Optional infrastructure override; callers select the transport with useEdge. */ edgeUrl?: string; } /** Lightweight transport shared by Codex and clients with their own GraphQL documents. */ -export function createRealtimeClient( - options: RealtimeClientOptions, -): RealtimeClient { - const { edgeUrl = DEFAULT_EDGE_REALTIME_URL, ...wsOptions } = options; +export function createEdgeClient(options: EdgeClientOptions): EdgeClient { + const { edgeUrl = DEFAULT_EDGE_URL, ...wsOptions } = options; type EventHandlers = NonNullable; const initialHandlers = wsOptions.on ? [wsOptions.on] : []; const mainHandlers = new Set(initialHandlers); @@ -77,7 +75,7 @@ export function createRealtimeClient( }; }, subscribe: (payload, sink, subscriptionOptions = {}) => { - if (disposed) throw new Error("Realtime client has been disposed"); + if (disposed) throw new Error("Edge client has been disposed"); if (!subscriptionOptions.useEdge) return main.subscribe(payload, sink); // Reject before creating an edge client or opening any upstream operation. const snapshot: SubscribePayload = JSON.parse( diff --git a/src/sdk/index.ts b/src/sdk/index.ts index a2bb464..e3d3ae4 100644 --- a/src/sdk/index.ts +++ b/src/sdk/index.ts @@ -3,14 +3,10 @@ import { GraphQLClient, Variables } from "graphql-request"; import { ExecutionResult, Sink } from "graphql-ws"; import WebSocket from "isomorphic-ws"; +import { createEdgeClient, EdgeClient, SubscriptionOptions } from "./edge"; import { invariant } from "./invariant"; import { Mutation } from "./Mutation"; import { Query } from "./Query"; -import { - createRealtimeClient, - RealtimeClient, - SubscriptionOptions, -} from "./realtime"; import { Subscribe } from "./Subscribe"; export type CleanupFunction = () => void; @@ -23,7 +19,7 @@ export type ApiConfig = { apiRealtimeUrl: string; // Optional deployment override for the edge transport. - apiEdgeRealtimeUrl?: string; + apiEdgeUrl?: string; // true if you want to connect a websocket, sometimes useful to turn this off in certain environments ws?: boolean; @@ -49,7 +45,7 @@ const defaultConfig: ApiConfig = { export class Codex { private client: GraphQLClient; - private wsClient: RealtimeClient | undefined; + private wsClient: EdgeClient | undefined; public queries: Query; public mutations: Mutation; public subscriptions: Subscribe; @@ -95,11 +91,11 @@ export class Codex { private createWebsocketClient(config: ApiConfig) { return config.ws - ? createRealtimeClient({ + ? createEdgeClient({ webSocketImpl: WebSocket, keepAlive: 10_000, // ping server every 10 seconds url: config.apiRealtimeUrl, - edgeUrl: config.apiEdgeRealtimeUrl, + edgeUrl: config.apiEdgeUrl, connectionParams: { Authorization: this.apiKey, ...config.wsHeaders, diff --git a/tsconfig.json b/tsconfig.json index 6101985..7bdc9c7 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -28,7 +28,7 @@ "include": [ "src/sdk/**/*.ts", "src/index.ts", - "src/realtime.ts", + "src/edge.ts", "!src/sdk/**/*.test.ts" ] } From 42bb5abfe78291032c34041325eb2431a77e0757 Mon Sep 17 00:00:00 2001 From: Kyle Poole Date: Wed, 14 Oct 2026 11:11:54 -0700 Subject: [PATCH 5/6] refactor(sdk): create edge transport only for opt-in subscriptions --- src/edge.ts | 9 +++- src/sdk/edge.integration.test.ts | 51 ++++++++++++++++++--- src/sdk/edge.test.ts | 59 ++++++++++++++++++++----- src/sdk/edge.ts | 61 ++++++++++++++++++------- src/sdk/index.test.ts | 1 - src/sdk/index.ts | 76 +++++++++++++++++++++----------- 6 files changed, 198 insertions(+), 59 deletions(-) diff --git a/src/edge.ts b/src/edge.ts index f468baf..9d524db 100644 --- a/src/edge.ts +++ b/src/edge.ts @@ -1,6 +1,11 @@ -export { createEdgeClient, DEFAULT_EDGE_URL } from "./sdk/edge"; +export { + createClient, + createEdgeClient, + DEFAULT_EDGE_URL, + withEdge, +} from "./sdk/edge"; export type { - EdgeClient, + ClientWithEdge, EdgeClientOptions, SubscriptionOptions, } from "./sdk/edge"; diff --git a/src/sdk/edge.integration.test.ts b/src/sdk/edge.integration.test.ts index 6aa1a2a..504aa21 100644 --- a/src/sdk/edge.integration.test.ts +++ b/src/sdk/edge.integration.test.ts @@ -1,10 +1,15 @@ import { once } from "node:events"; import type { AddressInfo } from "node:net"; -import { parseMessage } from "graphql-ws"; +import { type ClientOptions, createClient, parseMessage } from "graphql-ws"; import WebSocket from "isomorphic-ws"; -import { createEdgeClient } from "./edge"; +import { withEdge } from "./edge"; +import { Codex } from "./index"; + +function createClientWithEdge(options: ClientOptions & { edgeUrl?: string }) { + return withEdge(createClient(options), options); +} it("delivers edge and normal results over separate authenticated sockets", async () => { const requests: Array<{ source: string; type: string; payload: unknown }> = @@ -43,7 +48,7 @@ it("delivers edge and normal results over separate authenticated sockets", async const urls = servers.map( (server) => `ws://127.0.0.1:${(server.address() as AddressInfo).port}`, ); - const client = createEdgeClient({ + const client = createClientWithEdge({ url: urls[0], edgeUrl: urls[1], webSocketImpl: WebSocket, @@ -132,6 +137,42 @@ const edgeQuery = `subscription Events($input: OnTokenEventsCreatedInput!, $leve onTokenEventsCreated(input: $input, commitmentLevel: $levels) { events { transactionHash } } }`; +it("keeps SDK subscriptions on main until a valid edge opt-in and disposes both sockets", async () => { + const main = await subscriptionServer(); + const edge = await subscriptionServer(); + const sdk = new Codex("customer", { + apiRealtimeUrl: main.url, + apiEdgeUrl: edge.url, + }); + const variables = { input: { networkId: 1399811149 }, levels: ["Processed"] }; + const expected = JSON.parse(JSON.stringify({ query: edgeQuery, variables })); + const sink = { next: () => {}, error: () => {}, complete: () => {} }; + try { + const mainInit = once(main.server, "init"); + const mainRequest = once(main.server, "subscription"); + sdk.subscribe(edgeQuery, variables, sink); + expect((await mainInit)[0]).toEqual({ Authorization: "customer" }); + expect((await mainRequest)[0]).toEqual(expected); + expect(edge.server.clients.size).toBe(0); + expect(() => + sdk.subscribe(edgeQuery, { ...variables, levels: ["Confirmed"] }, sink, { + useEdge: true, + }), + ).toThrow("useEdge requires"); + expect(edge.server.clients.size).toBe(0); + const edgeInit = once(edge.server, "init"); + const edgeRequest = once(edge.server, "subscription"); + sdk.subscribe(edgeQuery, variables, sink, { useEdge: true }); + variables.input.networkId = 1; + variables.levels[0] = "Confirmed"; + expect((await edgeInit)[0]).toEqual({ Authorization: "customer" }); + expect((await edgeRequest)[0]).toEqual(expected); + } finally { + await sdk.dispose(); + await Promise.all([main.close(), edge.close()]); + } +}); + it("preserves the validated payload through delayed authentication and retries", async () => { const edge = await subscriptionServer(); let authenticate!: () => void; @@ -139,7 +180,7 @@ it("preserves the validated payload through delayed authentication and retries", authenticate = resolve; }); let serializedLevels = 0; - const client = createEdgeClient({ + const client = createClientWithEdge({ url: "ws://main.example/graphql", edgeUrl: edge.url, webSocketImpl: WebSocket, @@ -193,7 +234,7 @@ it("forwards edge events, supports custom pong handling, and terminates both soc const edge = await subscriptionServer(); const configuredConnected = jest.fn(); let activeSocket: WebSocket; - const client = createEdgeClient({ + const client = createClientWithEdge({ url: main.url, edgeUrl: edge.url, webSocketImpl: WebSocket, diff --git a/src/sdk/edge.test.ts b/src/sdk/edge.test.ts index d433d93..1efbc8e 100644 --- a/src/sdk/edge.test.ts +++ b/src/sdk/edge.test.ts @@ -1,12 +1,17 @@ import { type ClientOptions, createClient } from "graphql-ws"; -import { createEdgeClient, DEFAULT_EDGE_URL } from "./edge"; +import { createEdgeClient, DEFAULT_EDGE_URL, withEdge } from "./edge"; +import * as edgeModule from "./edge"; import { assertEdgeSubscription } from "./edgeSubscription"; import { EventCommitmentLevel } from "./generated/graphql"; import { Codex } from "./index"; jest.mock("graphql-ws", () => ({ createClient: jest.fn() })); +function createClientWithEdge(options: ClientOptions & { edgeUrl?: string }) { + return withEdge(createClient(options), options); +} + const transports: Array<{ options: ClientOptions; subscribe: jest.Mock; @@ -37,8 +42,42 @@ beforeEach(() => { }); }); +afterEach(() => { + jest.restoreAllMocks(); +}); + +it("creates a standalone edge transport without creating a main transport", () => { + createEdgeClient({ connectionParams: { Authorization: "customer" } }); + expect(transports).toHaveLength(1); + expect(transports[0].options.url).toBe(DEFAULT_EDGE_URL); +}); + +it("creates the SDK edge client only for a valid opt-in and reuses it", async () => { + const createEdge = jest.spyOn(edgeModule, "createEdgeClient"); + const sdk = new Codex("customer", { + apiEdgeUrl: "wss://edge.staging.example/graphql", + }); + sdk.subscribe(events, variables, sink); + expect(createEdge).not.toHaveBeenCalled(); + expect(() => + sdk.subscribe(events, { ...variables, levels: ["Confirmed"] }, sink, { + useEdge: true, + }), + ).toThrow("useEdge requires"); + expect(createEdge).not.toHaveBeenCalled(); + sdk.subscribe(events, variables, sink, { useEdge: true }); + sdk.subscribe(events, variables, sink, { useEdge: true }); + expect(createEdge).toHaveBeenCalledTimes(1); + expect(transports[1].options.url).toBe("wss://edge.staging.example/graphql"); + await sdk.dispose(); + expect(() => + sdk.subscribe(events, variables, sink, { useEdge: true }), + ).toThrow("disposed"); + expect(createEdge).toHaveBeenCalledTimes(1); +}); + it("reuses one edge transport while keeping ordinary subscriptions on the main transport", () => { - const client = createEdgeClient({ + const client = createClientWithEdge({ url: "wss://main.example/graphql", connectionParams: { Authorization: "customer" }, }); @@ -65,7 +104,7 @@ it("reuses one edge transport while keeping ordinary subscriptions on the main t it("uses current connection credentials and infrastructure options on both transports", async () => { let token = "first"; const connectionParams = async () => ({ Authorization: token }); - const client = createEdgeClient({ + const client = createClientWithEdge({ url: "wss://main.example/graphql", edgeUrl: "wss://edge.staging.example/graphql", connectionParams, @@ -96,7 +135,7 @@ it.each([ ])( "accepts eligible literal and default-variable subscriptions: %s", (query) => { - const client = createEdgeClient({ url: "wss://main.example/graphql" }); + const client = createClientWithEdge({ url: "wss://main.example/graphql" }); client.subscribe({ query }, sink, { useEdge: true }); expect(transports[1].subscribe).toHaveBeenCalledTimes(1); }, @@ -119,7 +158,7 @@ it.each([ ])( "rejects ineligible variables without sending an operation: %j", (override) => { - const client = createEdgeClient({ url: "wss://main.example/graphql" }); + const client = createClientWithEdge({ url: "wss://main.example/graphql" }); // An empty override exercises missing variables rather than the valid base. const values = Object.keys(override).length ? { ...variables, ...override } @@ -146,7 +185,7 @@ it.each([ `subscription { onEventsCreated(networkId: 1399811149, commitmentLevel: [Processed]) @skip(if: true) { id } }`, `subscription { ...Root } fragment Root on Subscription { onEventsCreated(networkId: 1399811149, commitmentLevel: [Processed]) { id } }`, ])("rejects ineligible documents: %s", (query) => { - const client = createEdgeClient({ url: "wss://main.example/graphql" }); + const client = createClientWithEdge({ url: "wss://main.example/graphql" }); expect(() => client.subscribe({ query }, sink, { useEdge: true })).toThrow( "useEdge requires", ); @@ -154,7 +193,7 @@ it.each([ }); it("rejects implicit pair quote selection instead of letting edge relay it through Oregon", () => { - const client = createEdgeClient({ url: "wss://main.example/graphql" }); + const client = createClientWithEdge({ url: "wss://main.example/graphql" }); expect(() => client.subscribe( { @@ -168,7 +207,7 @@ it("rejects implicit pair quote selection instead of letting edge relay it throu }); it("validates the selected operation in a multi-operation document", () => { - const client = createEdgeClient({ url: "wss://main.example/graphql" }); + const client = createClientWithEdge({ url: "wss://main.example/graphql" }); const query = `${events}\nsubscription Other { onPriceUpdated(networkId: 1) { priceUsd } }`; client.subscribe({ query, variables, operationName: "Events" }, sink, { useEdge: true, @@ -179,7 +218,7 @@ it("validates the selected operation in a multi-operation document", () => { }); it("disposes both transports and prevents subscriptions from recreating them", async () => { - const client = createEdgeClient({ url: "wss://main.example/graphql" }); + const client = createClientWithEdge({ url: "wss://main.example/graphql" }); client.subscribe({ query: events, variables }, sink, { useEdge: true }); await client.dispose(); expect(transports.map(({ dispose }) => dispose.mock.calls.length)).toEqual([ @@ -223,7 +262,7 @@ it("rejects sparse commitment arrays even before wire serialization", () => { }); it("validates serialized variables rather than a misleading toJSON value", () => { - const client = createEdgeClient({ url: "wss://main.example/graphql" }); + const client = createClientWithEdge({ url: "wss://main.example/graphql" }); expect(() => client.subscribe( { diff --git a/src/sdk/edge.ts b/src/sdk/edge.ts index be0247c..426f7a0 100644 --- a/src/sdk/edge.ts +++ b/src/sdk/edge.ts @@ -11,6 +11,8 @@ import { import { assertEdgeSubscription } from "./edgeSubscription"; +export { createClient }; + export const DEFAULT_EDGE_URL = "wss://fra.edge.codex.io/graphql"; export interface SubscriptionOptions { @@ -18,7 +20,7 @@ export interface SubscriptionOptions { useEdge?: boolean; } -export interface EdgeClient extends Client { +export interface ClientWithEdge extends Client { /** Observe both transports, or select one with useEdge. */ on( event: E, @@ -32,19 +34,40 @@ export interface EdgeClient extends Client { ): () => void; } -export interface EdgeClientOptions extends ClientOptions { - /** Optional infrastructure override; callers select the transport with useEdge. */ - edgeUrl?: string; +export interface EdgeClientOptions extends Omit { + /** Defaults to the regional edge endpoint. */ + url?: ClientOptions["url"]; +} + +/** Create a transport that connects only to edge. */ +export function createEdgeClient(options: EdgeClientOptions): Client { + return createClient({ ...options, url: options.url ?? DEFAULT_EDGE_URL }); +} + +/** Snapshot and validate the arguments that will actually be sent to edge. */ +export function prepareEdgeSubscription( + payload: SubscribePayload, + replacer?: ClientOptions["jsonMessageReplacer"], +): SubscribePayload { + const snapshot: SubscribePayload = JSON.parse( + JSON.stringify(payload, replacer), + ); + assertEdgeSubscription(snapshot); + return snapshot; } -/** Lightweight transport shared by Codex and clients with their own GraphQL documents. */ -export function createEdgeClient(options: EdgeClientOptions): EdgeClient { +/** Add per-subscription edge routing to an existing GraphQL client. */ +export function withEdge( + main: Client, + options: ClientOptions & { edgeUrl?: string }, +): ClientWithEdge { const { edgeUrl = DEFAULT_EDGE_URL, ...wsOptions } = options; type EventHandlers = NonNullable; - const initialHandlers = wsOptions.on ? [wsOptions.on] : []; - const mainHandlers = new Set(initialHandlers); - const edgeHandlers = new Set(initialHandlers); - const handlersFor = (useEdge: boolean): EventHandlers => { + const mainHandlers = new Set(); + const edgeHandlers = new Set( + wsOptions.on ? [wsOptions.on] : [], + ); + const handlersFor = (useEdge: boolean): Required => { const handlers = () => Array.from(useEdge ? edgeHandlers : mainHandlers); // Install before createClient so even non-lazy connection events are observed. return { @@ -60,7 +83,15 @@ export function createEdgeClient(options: EdgeClientOptions): EdgeClient { error: (...args) => handlers().forEach((on) => on?.error?.(...args)), }; }; - const main = createClient({ ...wsOptions, on: handlersFor(false) }); + const mainEvents = handlersFor(false); + main.on("connecting", mainEvents.connecting); + main.on("opened", mainEvents.opened); + main.on("connected", mainEvents.connected); + main.on("ping", mainEvents.ping); + main.on("pong", mainEvents.pong); + main.on("message", mainEvents.message); + main.on("closed", mainEvents.closed); + main.on("error", mainEvents.error); let edge: Client | undefined; let disposed = false; return { @@ -78,15 +109,15 @@ export function createEdgeClient(options: EdgeClientOptions): EdgeClient { if (disposed) throw new Error("Edge client has been disposed"); if (!subscriptionOptions.useEdge) return main.subscribe(payload, sink); // Reject before creating an edge client or opening any upstream operation. - const snapshot: SubscribePayload = JSON.parse( - JSON.stringify(payload, wsOptions.jsonMessageReplacer), + const snapshot = prepareEdgeSubscription( + payload, + wsOptions.jsonMessageReplacer, ); - assertEdgeSubscription(snapshot); if (!edge) { // Edge payloads are serialized once, before validation. Preserve custom // serialization for handshakes without transforming subscriptions again. let subscriptionMessage = false; - edge = createClient({ + edge = createEdgeClient({ ...wsOptions, url: edgeUrl, on: handlersFor(true), diff --git a/src/sdk/index.test.ts b/src/sdk/index.test.ts index 0f67196..e8a664c 100644 --- a/src/sdk/index.test.ts +++ b/src/sdk/index.test.ts @@ -448,7 +448,6 @@ describe("Codex", () => { variables, }, sink, - undefined, ); }); }); diff --git a/src/sdk/index.ts b/src/sdk/index.ts index e3d3ae4..cfa1757 100644 --- a/src/sdk/index.ts +++ b/src/sdk/index.ts @@ -1,9 +1,18 @@ import { TypedDocumentNode } from "@graphql-typed-document-node/core"; import { GraphQLClient, Variables } from "graphql-request"; -import { ExecutionResult, Sink } from "graphql-ws"; +import { + Client as GraphQLWsClient, + createClient, + ExecutionResult, + Sink, +} from "graphql-ws"; import WebSocket from "isomorphic-ws"; -import { createEdgeClient, EdgeClient, SubscriptionOptions } from "./edge"; +import { + createEdgeClient, + prepareEdgeSubscription, + SubscriptionOptions, +} from "./edge"; import { invariant } from "./invariant"; import { Mutation } from "./Mutation"; import { Query } from "./Query"; @@ -45,7 +54,9 @@ const defaultConfig: ApiConfig = { export class Codex { private client: GraphQLClient; - private wsClient: EdgeClient | undefined; + private wsClient: GraphQLWsClient | undefined; + private edgeWsClient: GraphQLWsClient | undefined; + private disposed = false; public queries: Query; public mutations: Mutation; public subscriptions: Subscribe; @@ -89,19 +100,20 @@ export class Codex { }); } + private websocketOptions(config: ApiConfig) { + return { + webSocketImpl: WebSocket, + keepAlive: 10_000, // ping server every 10 seconds + url: config.apiRealtimeUrl, + connectionParams: { + Authorization: this.apiKey, + ...config.wsHeaders, + }, + }; + } + private createWebsocketClient(config: ApiConfig) { - return config.ws - ? createEdgeClient({ - webSocketImpl: WebSocket, - keepAlive: 10_000, // ping server every 10 seconds - url: config.apiRealtimeUrl, - edgeUrl: config.apiEdgeUrl, - connectionParams: { - Authorization: this.apiKey, - ...config.wsHeaders, - }, - }) - : undefined; + return config.ws ? createClient(this.websocketOptions(config)) : undefined; } private async getRequestHeaders() { @@ -122,7 +134,9 @@ export class Codex { */ public updateConfig(incomingConfig: Partial) { // If we have an existing websocket connection, dispose of it. - if (this.wsClient) void this.dispose(); + if (this.wsClient || this.edgeWsClient) void this.dispose(); + this.edgeWsClient = undefined; + this.disposed = false; this.apiConfig = { ...this.apiConfig, ...incomingConfig }; // Merge the incoming config with the existing config. @@ -152,7 +166,8 @@ export class Codex { * It's useful for when you need to modify some of the configuration, or when you're done with the class. */ public async dispose() { - await this.wsClient?.dispose(); + this.disposed = true; + await Promise.all([this.wsClient?.dispose(), this.edgeWsClient?.dispose()]); } /** @@ -229,14 +244,23 @@ export class Codex { throw new Error("Websocket is not enabled in the config"); } - const cleanup = this.wsClient.subscribe( - { - query: doc, - variables: args, - }, - sink, - options, - ); - return cleanup; + if (this.disposed) throw new Error("SDK has been disposed"); + const payload = { query: doc, variables: args }; + if (!options?.useEdge) + return this.wsClient.subscribe(payload, sink); + + // Validate before creating the edge transport or starting an operation. + const snapshot = prepareEdgeSubscription(payload); + if (!this.edgeWsClient) { + const config = this.parseConfig(this.apiConfig ?? {}); + this.edgeWsClient = createEdgeClient({ + ...this.websocketOptions(config), + url: config.apiEdgeUrl, + }); + this.edgeWsClient.on("error", (error) => { + console.error("Websocket error: ", error); + }); + } + return this.edgeWsClient.subscribe(snapshot, sink); } } From 825aafbe8a99c31c59226800fdbd31b1b0c2851c Mon Sep 17 00:00:00 2001 From: Kyle Poole Date: Wed, 14 Oct 2026 12:51:08 -0700 Subject: [PATCH 6/6] fix(sdk): accept mixed edge commitments containing Processed --- src/sdk/edge.integration.test.ts | 11 +++++++---- src/sdk/edge.test.ts | 30 +++++++++++++++++++++++++----- src/sdk/edge.ts | 2 +- src/sdk/edgeSubscription.ts | 12 ++++++++++-- 4 files changed, 43 insertions(+), 12 deletions(-) diff --git a/src/sdk/edge.integration.test.ts b/src/sdk/edge.integration.test.ts index 504aa21..bd473eb 100644 --- a/src/sdk/edge.integration.test.ts +++ b/src/sdk/edge.integration.test.ts @@ -54,7 +54,7 @@ it("delivers edge and normal results over separate authenticated sockets", async webSocketImpl: WebSocket, connectionParams: { Authorization: "test-customer" }, }); - const query = `subscription { onEventsCreated(id: "pair:1399811149", commitmentLevel: [Processed]) { events { transactionHash } } }`; + const query = `subscription { onEventsCreated(id: "pair:1399811149", commitmentLevel: [Processed, Confirmed]) { events { transactionHash } } }`; try { expect(() => client.subscribe( @@ -144,7 +144,10 @@ it("keeps SDK subscriptions on main until a valid edge opt-in and disposes both apiRealtimeUrl: main.url, apiEdgeUrl: edge.url, }); - const variables = { input: { networkId: 1399811149 }, levels: ["Processed"] }; + const variables = { + input: { networkId: 1399811149 }, + levels: ["Processed", "Confirmed"], + }; const expected = JSON.parse(JSON.stringify({ query: edgeQuery, variables })); const sink = { next: () => {}, error: () => {}, complete: () => {} }; try { @@ -199,7 +202,7 @@ it("preserves the validated payload through delayed authentication and retries", }); const variables = { input: { networkId: 1399811149, tokenAddress: "original" }, - levels: ["Processed"], + levels: ["Preprocessed", "Processed", "Confirmed"], }; const payload = { query: edgeQuery, variables }; const expected = JSON.parse(JSON.stringify(payload)); @@ -211,7 +214,7 @@ it("preserves the validated payload through delayed authentication and retries", { next: () => {}, error: () => {}, complete: () => {} }, { useEdge: true }, ); - variables.levels[0] = "Confirmed"; + variables.levels.splice(0, variables.levels.length, "Confirmed"); variables.input.networkId = 1; variables.input.tokenAddress = "changed"; payload.query = "subscription { onPriceUpdated { priceUsd } }"; diff --git a/src/sdk/edge.test.ts b/src/sdk/edge.test.ts index 1efbc8e..5ef6bfd 100644 --- a/src/sdk/edge.test.ts +++ b/src/sdk/edge.test.ts @@ -132,6 +132,11 @@ it.each([ `subscription { onTokenBarsUpdated(tokenId: "token:1399811149", commitmentLevel: [Processed]) { timestamp } }`, `subscription { onTokenBarsUpdated(networkId: 1399811149, commitmentLevel: [Processed]) { timestamp } }`, `subscription($network: Int = 1399811149, $levels: [EventCommitmentLevel!] = [Processed]) { onEventsCreated(networkId: $network, commitmentLevel: $levels) { events { transactionHash } } }`, + `subscription { onEventsCreated(id: "pair:1399811149", commitmentLevel: [Confirmed, Processed]) { events { transactionHash } } }`, + `subscription { onTokenEventsCreated(input: {networkId: 1399811149}, commitmentLevel: [Preprocessed, Processed, Confirmed]) { events { transactionHash } } }`, + `subscription { onBarsUpdated(pairId: "pair:1399811149", quoteToken: token0, commitmentLevel: [Processed, Confirmed]) { timestamp } }`, + `subscription { onTokenBarsUpdated(tokenId: "token:1399811149", commitmentLevel: [Preprocessed, Processed]) { timestamp } }`, + `subscription($network: Int = 1399811149, $levels: [EventCommitmentLevel!] = [Confirmed, Processed]) { onEventsCreated(networkId: $network, commitmentLevel: $levels) { events { transactionHash } } }`, ])( "accepts eligible literal and default-variable subscriptions: %s", (query) => { @@ -150,8 +155,10 @@ it.each([ { levels: Object.assign(new Array(2), { 0: "Processed" }) }, { levels: ["Confirmed"] }, { levels: ["Preprocessed"] }, - { levels: ["Processed", "Confirmed"] }, - { levels: ["Processed", "Preprocessed"] }, + { levels: ["Preprocessed", "Confirmed"] }, + { levels: ["Processed", "Finalized"] }, + { levels: ["Processed", null] }, + { levels: ["Processed", 1] }, { input: { networkId: 8453 } }, { input: { networkId: "1399811149" } }, { input: { tokenAddress: "token" } }, @@ -229,19 +236,32 @@ it("disposes both transports and prevents subscriptions from recreating them", a ).toThrow("disposed"); }); -it("routes both raw and generated SDK subscriptions through useEdge", async () => { +it("routes raw and generated mixed SDK subscriptions through useEdge without dropping commitments", async () => { const sdk = new Codex("customer"); - sdk.subscribe(events, variables, sink, { useEdge: true }); + const levels = [ + EventCommitmentLevel.Preprocessed, + EventCommitmentLevel.Processed, + EventCommitmentLevel.Confirmed, + ]; + const mixedVariables = { ...variables, levels }; + sdk.subscribe(events, mixedVariables, sink, { useEdge: true }); sdk.subscriptions.onTokenEventsCreated( { input: variables.input, - commitmentLevel: [EventCommitmentLevel.Processed], + commitmentLevel: levels, }, sink, { useEdge: true }, ); expect(transports).toHaveLength(2); expect(transports[1].subscribe).toHaveBeenCalledTimes(2); + expect(transports[1].subscribe.mock.calls[0][0].variables).toEqual( + mixedVariables, + ); + expect(transports[1].subscribe.mock.calls[1][0].variables).toEqual({ + input: variables.input, + commitmentLevel: levels, + }); sdk.updateConfig({ wsHeaders: { Authorization: "new-customer" } }); sdk.subscribe(events, variables, sink, { useEdge: true }); expect(transports[3].options.connectionParams).toEqual({ diff --git a/src/sdk/edge.ts b/src/sdk/edge.ts index 426f7a0..8c83831 100644 --- a/src/sdk/edge.ts +++ b/src/sdk/edge.ts @@ -16,7 +16,7 @@ export { createClient }; export const DEFAULT_EDGE_URL = "wss://fra.edge.codex.io/graphql"; export interface SubscriptionOptions { - /** Experimental: deliver supported Processed Solana streams directly from edge. */ + /** Experimental: use edge for supported Solana streams whose commitments include Processed. */ useEdge?: boolean; } diff --git a/src/sdk/edgeSubscription.ts b/src/sdk/edgeSubscription.ts index 30db689..f545542 100644 --- a/src/sdk/edgeSubscription.ts +++ b/src/sdk/edgeSubscription.ts @@ -13,7 +13,7 @@ const SUPPORTED_OPERATIONS = new Set([ export function assertEdgeSubscription(payload: SubscribePayload): void { function fail(): never { throw new Error( - "useEdge requires a supported Solana subscription with only Processed commitmentLevel", + "useEdge requires a supported Solana subscription with commitmentLevel including Processed", ); } const document = parse(payload.query); @@ -47,7 +47,15 @@ export function assertEdgeSubscription(payload: SubscribePayload): void { const levels = Array.isArray(args.commitmentLevel) ? args.commitmentLevel : [args.commitmentLevel]; - if (!levels.length || [...levels].some((level) => level !== "Processed")) + if ( + !levels.includes("Processed") || + [...levels].some( + (level) => + level !== "Processed" && + level !== "Confirmed" && + level !== "Preprocessed", + ) + ) fail(); const networks: unknown[] = [];