diff --git a/packages/js-client-rest/README.md b/packages/js-client-rest/README.md index bdd3a15..66c277f 100644 --- a/packages/js-client-rest/README.md +++ b/packages/js-client-rest/README.md @@ -84,7 +84,20 @@ try { ## Support -The REST implementation relies on the native [fetch API](https://developer.mozilla.org/en-US/docs/Web/API/Fetch_API), which is available in Deno and Node.js (starting on v18.0.0 without experimental flag). The Deno implementation [supports HTTP/2](https://deno.com/blog/every-web-api-in-deno#fetch-request-response-and-headers) whereas Node.js is still lagging on the spec and provide only HTTP 1.1 support (this is due to the fact that under the hood Node.js still relies on [undici](https://github.com/nodejs/undici)). +The REST implementation relies on the native [fetch API](https://developer.mozilla.org/en-US/docs/Web/API/Fetch_API), which is available in Deno and Node.js (starting on v18.0.0 without experimental flag). This package does not install its own fetch implementation or dispatcher layer, so transport behavior is delegated to the host runtime. + +If you need custom transport behavior, you can inject your own `fetch` implementation when constructing the client: + +```ts +import {QdrantClient} from '@qdrant/js-client-rest'; + +const client = new QdrantClient({ + url: 'http://localhost:6333', + fetch: globalThis.fetch, +}); +``` + +This also allows advanced Node.js setups to keep using user-managed transport tooling such as `undici` without making it a dependency of this package. ## Releases diff --git a/packages/js-client-rest/package.json b/packages/js-client-rest/package.json index 7e5e47f..4a17d60 100644 --- a/packages/js-client-rest/package.json +++ b/packages/js-client-rest/package.json @@ -54,8 +54,7 @@ "openapi_schema_remote": "https://raw.githubusercontent.com/qdrant/qdrant/dev/docs/redoc/master/openapi.json" }, "dependencies": { - "@qdrant/openapi-typescript-fetch": "1.2.6", - "undici": "^6.23.0" + "@qdrant/openapi-typescript-fetch": "1.2.6" }, "devDependencies": { "@rollup/plugin-commonjs": "^24.1.0", diff --git a/packages/js-client-rest/src/api-client.ts b/packages/js-client-rest/src/api-client.ts index 404796b..f32c862 100644 --- a/packages/js-client-rest/src/api-client.ts +++ b/packages/js-client-rest/src/api-client.ts @@ -1,6 +1,6 @@ -import {ApiError, Fetcher, Middleware} from '@qdrant/openapi-typescript-fetch'; +import {ApiError, Middleware} from '@qdrant/openapi-typescript-fetch'; import {paths} from './openapi/generated_schema.js'; -import {createDispatcher} from './dispatcher.js'; +import {createFetcher} from './fetcher.js'; import { QdrantClientResourceExhaustedError, QdrantClientTimeoutError, @@ -10,7 +10,7 @@ import {RestArgs} from './types.js'; import {createClientApi} from './openapi/generated_api_client.js'; import {ClientApi} from './openapi/generated_client_type.js'; -export type Client = ReturnType>; +export type Client = ReturnType>; export function createApis(baseUrl: string, args: RestArgs): ClientApi { const client = createClient(baseUrl, args); @@ -19,7 +19,7 @@ export function createApis(baseUrl: string, args: RestArgs): ClientApi { export type OpenApiClient = ReturnType; -export function createClient(baseUrl: string, {headers, timeout, connections}: RestArgs): Client { +export function createClient(baseUrl: string, {headers, timeout, fetch}: RestArgs): Client { const use: Middleware[] = []; if (Number.isFinite(timeout)) { use.push(async (url, init, next) => { @@ -48,7 +48,7 @@ export function createClient(baseUrl: string, {headers, timeout, connections}: R } } catch (error) { if (error instanceof ApiError && error.status === 429) { - const retryAfterHeader = error.headers.get('retry-after')?.[0]; + const retryAfterHeader = error.headers.get('retry-after'); if (retryAfterHeader) { throw new QdrantClientResourceExhaustedError(error.message, retryAfterHeader); } @@ -59,20 +59,13 @@ export function createClient(baseUrl: string, {headers, timeout, connections}: R throw QdrantClientUnexpectedResponseError.forResponse(response); }); - const client = Fetcher.for(); - // Configure client with 'undici' agent which is used in Node 18+ + const client = createFetcher(); client.configure({ baseUrl, init: { headers, - // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment - dispatcher: - typeof process !== 'undefined' && - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - process.versions?.node - ? createDispatcher(connections) - : undefined, }, + fetch, use, }); diff --git a/packages/js-client-rest/src/dispatcher.ts b/packages/js-client-rest/src/dispatcher.ts deleted file mode 100644 index 12edfd1..0000000 --- a/packages/js-client-rest/src/dispatcher.ts +++ /dev/null @@ -1,23 +0,0 @@ -/* eslint-disable @typescript-eslint/no-unsafe-call, @typescript-eslint/no-unsafe-return */ - -import {Agent} from 'undici'; - -declare global { - interface RequestInit { - dispatcher?: Agent | undefined; - } -} - -export const createDispatcher = (connections = 25) => - // eslint-disable-next-line @typescript-eslint/no-unsafe-return - new Agent({ - // timeouts are handled by AbortSignal in our middleware - bodyTimeout: 0, - headersTimeout: 0, - // https://stackoverflow.com/a/36437932/558180 - // pipelining: 1, - // a sensible max connections value - connections, - // will be overriden by header Keep-Alive, just as sensible default - keepAliveTimeout: 10_000, - }); diff --git a/packages/js-client-rest/src/fetcher.ts b/packages/js-client-rest/src/fetcher.ts new file mode 100644 index 0000000..013fe97 --- /dev/null +++ b/packages/js-client-rest/src/fetcher.ts @@ -0,0 +1,326 @@ +import {ApiError, Middleware, TypedFetch} from '@qdrant/openapi-typescript-fetch'; + +let bigintReviver: ((this: unknown, key: string, value: unknown, context: {source: string}) => unknown) | undefined; +let bigintReplacer: ((this: unknown, key: string, value: unknown) => unknown) | undefined; + +if ('rawJSON' in JSON) { + bigintReviver = function (_key, value, context) { + if (Number.isInteger(value) && !Number.isSafeInteger(value)) { + try { + return BigInt(context.source); + } catch { + return value; + } + } + + return value; + }; + + bigintReplacer = function (_key, value) { + if (typeof value === 'bigint') { + return JSON.rawJSON?.(String(value)) ?? String(value); + } + + return value; + }; +} + +type Method = 'get' | 'post' | 'put' | 'patch' | 'delete' | 'head' | 'options'; + +type OpenapiPaths = { + [P in keyof Paths]: { + [M in Method]?: unknown; + }; +}; + +type Client> = { + configure: (config: FetchConfig) => void; + path:

( + path: P, + ) => { + method: ( + method: M, + ) => { + create: (queryParams?: Record) => TypedFetch; + }; + }; +}; + +type ApiResponse = { + readonly headers: Headers; + readonly url: string; + readonly ok: boolean; + readonly status: number; + readonly statusText: string; + readonly data: R; +}; + +type CustomRequestInit = Omit & { + readonly headers: Headers; +}; + +type RuntimeFetch = (url: string, init: CustomRequestInit) => Promise; + +type RequestDefinition = { + baseUrl: string; + method: Method; + path: string; + queryParams: string[]; + payload: unknown; + init?: RequestInit; + fetch: RuntimeFetch; +}; + +export type FetchImplementation = typeof globalThis.fetch; + +type FetchConfig = { + baseUrl?: string; + init?: RequestInit; + use?: Middleware[]; + fetch?: FetchImplementation; +}; + +const canSendBody = (method: Method) => + method === 'post' || method === 'put' || method === 'patch' || method === 'delete'; + +function queryString(params: Record) { + const qs: string[] = []; + const encode = (key: string, value: unknown) => `${encodeURIComponent(key)}=${encodeURIComponent(String(value))}`; + + Object.keys(params).forEach((key) => { + const value = params[key]; + if (value != null) { + if (Array.isArray(value)) { + value.forEach((item) => qs.push(encode(key, item))); + } else { + qs.push(encode(key, value)); + } + } + }); + + return qs.length > 0 ? `?${qs.join('&')}` : ''; +} + +function omitKeys(payload: Record, keys: string[]) { + return Object.fromEntries(Object.entries(payload).filter(([key]) => !keys.includes(key))); +} + +function getPath(path: string, payload: Record) { + return path.replace(/\{([^}]+)\}/g, (_, key: string) => { + const value = encodeURIComponent(String(payload[key])); + return value; + }); +} + +function getQuery(method: Method, payload: Record, query: string[]) { + if (canSendBody(method)) { + return queryString(Object.fromEntries(query.map((key) => [key, payload[key]])) as Record); + } + + return queryString(payload); +} + +function getHeaders(body: BodyInit | undefined, init?: HeadersInit) { + const headers = new Headers(init); + + if (body !== undefined && !(body instanceof FormData) && !headers.has('Content-Type')) { + headers.append('Content-Type', 'application/json'); + } + + if (!headers.has('Accept')) { + headers.append('Accept', 'application/json'); + } + + return headers; +} + +function getBody(method: Method, payload: unknown) { + if (!canSendBody(method)) { + return undefined; + } + + const body = payload instanceof FormData ? payload : JSON.stringify(payload, bigintReplacer); + return method === 'delete' && body === '{}' ? undefined : body; +} + +function mergeRequestInit(first?: RequestInit, second?: RequestInit): RequestInit { + const headers = new Headers(first?.headers); + const other = new Headers(second?.headers); + + other.forEach((value, key) => { + headers.set(key, value); + }); + + return {...first, ...second, headers}; +} + +function isObjectRecord(payload: unknown): payload is Record { + if (!payload || typeof payload !== 'object') { + return false; + } + + return !Array.isArray(payload) && !(payload instanceof FormData); +} + +function clonePayload(payload: unknown): Record | unknown[] | FormData { + if (!payload || typeof payload !== 'object') { + return {}; + } + + if (payload instanceof FormData) { + return payload; + } + + return Object.assign(Array.isArray(payload) ? [] : {}, payload) as Record | unknown[]; +} + +function getFetchParams(request: RequestDefinition) { + const payload = clonePayload(request.payload); + const pathPayload = isObjectRecord(payload) ? payload : {}; + const pathParamKeys = Array.from(request.path.matchAll(/\{([^}]+)\}/g), ([, key]) => key); + const requestPayload = isObjectRecord(payload) ? omitKeys(pathPayload, pathParamKeys) : payload; + const path = getPath(request.path, pathPayload); + const query = isObjectRecord(requestPayload) ? getQuery(request.method, requestPayload, request.queryParams) : ''; + const body = + isObjectRecord(requestPayload) && canSendBody(request.method) + ? getBody(request.method, omitKeys(requestPayload, request.queryParams)) + : getBody(request.method, requestPayload); + const headers = canSendBody(request.method) + ? getHeaders(body, request.init?.headers) + : new Headers(request.init?.headers); + + return { + url: request.baseUrl + path + query, + init: { + ...request.init, + method: request.method.toUpperCase(), + headers, + body, + } satisfies CustomRequestInit, + }; +} + +async function getResponseData(response: Response) { + if (response.status === 204) { + return undefined; + } + + const contentType = response.headers.get('content-type'); + const responseText = await response.text(); + + if (contentType?.includes('application/json')) { + return JSON.parse( + responseText, + bigintReviver as ((this: unknown, key: string, value: unknown) => unknown) | undefined, + ); + } + + try { + return JSON.parse( + responseText, + bigintReviver as ((this: unknown, key: string, value: unknown) => unknown) | undefined, + ); + } catch { + return responseText; + } +} + +async function fetchJson(url: string, init: CustomRequestInit, fetchImpl: FetchImplementation): Promise { + const response = await fetchImpl(url, init); + const data = await getResponseData(response); + const result = { + headers: response.headers, + url: response.url, + ok: response.ok, + status: response.status, + statusText: response.statusText, + data, + }; + + if (result.ok) { + return result; + } + + throw new ApiError(result); +} + +function wrapMiddlewares(middlewares: Middleware[], fetch: RuntimeFetch): RuntimeFetch { + const handler = async (index: number, url: string, init: CustomRequestInit): Promise => { + if (index === middlewares.length) { + return fetch(url, init); + } + + const current = middlewares[index]; + return current(url, init, (nextUrl, nextInit) => handler(index + 1, nextUrl, nextInit)); + }; + + return (url, init) => handler(0, url, init); +} + +async function fetchUrl(request: RequestDefinition) { + const {url, init} = getFetchParams(request); + return request.fetch(url, init); +} + +function createTypedFetch(fetch: (payload: unknown, init?: RequestInit) => Promise) { + const fun = async (payload: unknown, init?: RequestInit) => { + try { + return await fetch(payload, init); + } catch (err) { + if (err instanceof ApiError) { + throw new fun.Error(err); + } + throw err; + } + }; + + fun.Error = class extends ApiError { + constructor(error: ApiError) { + super(error); + Object.setPrototypeOf(this, new.target.prototype); + } + + getActualType() { + return { + status: this.status, + data: this.data as unknown, + }; + } + }; + + return fun as unknown as TypedFetch; +} + +export function createFetcher>(): Client { + let baseUrl = ''; + let defaultInit: RequestInit = {}; + let fetchImpl: FetchImplementation = globalThis.fetch; + const middlewares: Middleware[] = []; + + return { + configure: (config) => { + baseUrl = config.baseUrl ?? ''; + defaultInit = config.init ?? {}; + fetchImpl = config.fetch ?? globalThis.fetch; + middlewares.splice(0, middlewares.length, ...(config.use ?? [])); + }, + path: (path) => ({ + method: (method) => ({ + create: (queryParams) => + createTypedFetch((payload, init) => + fetchUrl({ + baseUrl, + path: String(path), + method: String(method).toLowerCase() as Method, + queryParams: Object.keys(queryParams ?? {}), + payload, + init: mergeRequestInit(defaultInit, init), + fetch: wrapMiddlewares(middlewares, (url, requestInit) => + fetchJson(url, requestInit, fetchImpl), + ), + }), + ), + }), + }), + }; +} diff --git a/packages/js-client-rest/src/qdrant-client.ts b/packages/js-client-rest/src/qdrant-client.ts index 7c52e3a..ad9605e 100644 --- a/packages/js-client-rest/src/qdrant-client.ts +++ b/packages/js-client-rest/src/qdrant-client.ts @@ -3,6 +3,7 @@ import {QdrantClientConfigError} from './errors.js'; import {RestArgs, Schemas} from './types.js'; import {PACKAGE_VERSION, ClientVersion} from './client-version.js'; import {ClientApi} from './openapi/generated_client_type.js'; +import {FetchImplementation} from './fetcher.js'; const noResultError = (): never => { throw new Error('Result came uninitialized'); @@ -24,9 +25,13 @@ export type QdrantClientParams = { */ headers?: Record; /** - * The Node.js fetch API (undici) uses HTTP/1.1 under the hood. - * This indicates the maximum number of keep-alive connections - * to open simultaneously while building a request pool in memory. + * Custom fetch implementation used by the REST transport. + */ + fetch?: FetchImplementation; + /** + * Deprecated: native fetch is used directly, so connection pooling is + * managed by the runtime and this option currently has no effect. + * @deprecated */ maxConnections?: number; /** @@ -113,8 +118,7 @@ export class QdrantClient { const address = this._port ? `${this._host}:${this._port}` : this._host; this._restUri = `${this._scheme}://${address}${this._prefix}`; - const connections = args.maxConnections; - const restArgs: RestArgs = {headers, timeout, connections}; + const restArgs: RestArgs = {headers, timeout, fetch: args.fetch}; this._openApiClient = createApis(this._restUri, restArgs); diff --git a/packages/js-client-rest/src/types.ts b/packages/js-client-rest/src/types.ts index 3af7a79..95672ca 100644 --- a/packages/js-client-rest/src/types.ts +++ b/packages/js-client-rest/src/types.ts @@ -1,9 +1,10 @@ +import {FetchImplementation} from './fetcher.js'; import {components} from './openapi/generated_schema.js'; export interface RestArgs { headers: Headers; timeout: number; - connections?: number; + fetch?: FetchImplementation; } // Definitions (in OpenAPI 2.0) or Schemas (in OpenAPI 3.0) – Data models that describe your API inputs and outputs. diff --git a/packages/js-client-rest/tests/unit/api-client.test.ts b/packages/js-client-rest/tests/unit/api-client.test.ts index 3699a58..848cb72 100644 --- a/packages/js-client-rest/tests/unit/api-client.test.ts +++ b/packages/js-client-rest/tests/unit/api-client.test.ts @@ -1,17 +1,20 @@ +import {ApiError} from '@qdrant/openapi-typescript-fetch'; import {createApis, createClient} from '../../src/api-client.js'; -import {QdrantClientTimeoutError, QdrantClientUnexpectedResponseError} from '../../src/errors.js'; +import {QdrantClientResourceExhaustedError, QdrantClientTimeoutError} from '../../src/errors.js'; +import {createFetcher} from '../../src/fetcher.js'; import {vi, describe, test, expect, beforeEach, afterEach} from 'vitest'; describe('apiClient', () => { - const headers = new Headers(); - headers.set('content-type', 'application/json'); - const createFetchResponse = (status: number) => ({ - headers, - ok: true, + const createFetchResponse = (status: number, body: unknown = {error_message: 'response error'}) => ({ + headers: new Headers([['content-type', 'application/json']]), + ok: status >= 200 && status < 300, status, - json: () => new Promise((resolve) => resolve({error_message: 'response error'})), - text: () => new Promise((resolve) => resolve(JSON.stringify({error_message: 'response error'}))), + statusText: status === 200 ? 'OK' : status === 429 ? 'Too Many Requests' : 'Bad Request', + url: 'http://my-domain.com/test', + json: () => new Promise((resolve) => resolve(body)), + text: () => new Promise((resolve) => resolve(JSON.stringify(body))), }); + const headers = new Headers([['content-type', 'application/json']]); let originalFetch: typeof global.fetch; beforeEach(() => { @@ -40,6 +43,9 @@ describe('apiClient', () => { method: 'GET', }), ); + + const [, init] = vi.mocked(global.fetch).mock.calls[0] ?? []; + expect(init).not.toHaveProperty('dispatcher'); }); test('status 400', async () => { @@ -51,7 +57,23 @@ describe('apiClient', () => { }); const telemetry = client.path('/telemetry').method('get').create(); - await expect(telemetry({})).rejects.toThrowError(QdrantClientUnexpectedResponseError); + await expect(telemetry({})).rejects.toThrowError(ApiError); + }); + + test('status 429 preserves full retry-after value', async () => { + const response = createFetchResponse(429); + response.headers.set('retry-after', '10'); + global.fetch = vi.fn().mockResolvedValue(response); + + const client = createClient('http://my-domain.com', { + timeout: Infinity, + headers, + }); + const telemetry = client.path('/telemetry').method('get').create(); + + await expect(telemetry({})).rejects.toMatchObject({ + retry_after: 10, + } satisfies Partial); }); test('signal abort: timeout', async () => { @@ -67,4 +89,67 @@ describe('apiClient', () => { await expect(telemetry({})).rejects.toThrowError(QdrantClientTimeoutError); }); + + test('uses injected fetch implementation when provided', async () => { + const customFetch = vi.fn().mockResolvedValue(createFetchResponse(200)); + global.fetch = vi.fn().mockResolvedValue(createFetchResponse(400)); + + const apis = createApis('http://my-domain.com', { + timeout: Infinity, + headers, + fetch: customFetch, + }); + + await expect(apis.collectionExists({collection_name: 'my-collection'})).resolves.toMatchObject({ + data: {error_message: 'response error'}, + }); + + expect(customFetch).toHaveBeenCalledOnce(); + expect(global.fetch).not.toHaveBeenCalled(); + }); + + test('preserves FormData request bodies', async () => { + const client = createFetcher<{ + '/snapshots/upload': { + post: unknown; + }; + }>(); + const formData = new FormData(); + formData.set('snapshot', new Blob(['snapshot'])); + const customFetch = vi.fn().mockResolvedValue(createFetchResponse(200, {result: true})); + client.configure({ + baseUrl: 'http://my-domain.com', + fetch: customFetch, + }); + const uploadSnapshot = client.path('/snapshots/upload').method('post').create(); + + await uploadSnapshot(formData as never); + + const calls = customFetch.mock.calls as Array<[string, RequestInit]>; + const call = calls[0]; + expect(call).toBeDefined(); + const init = call?.[1]; + expect(init?.body).toBe(formData); + }); + + test('parses large JSON integers as bigint when runtime support is available', async () => { + const customFetch = vi.fn().mockResolvedValue({ + headers: new Headers([['content-type', 'application/json']]), + ok: true, + status: 200, + statusText: 'OK', + url: 'http://my-domain.com/test', + text: () => Promise.resolve('{"value":9223372036854775807}'), + }); + + const apis = createApis('http://my-domain.com', { + timeout: Infinity, + headers, + fetch: customFetch, + }); + + await expect(apis.collectionExists({collection_name: 'my-collection'})).resolves.toMatchObject({ + data: {value: BigInt('9223372036854775807')}, + }); + }); }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3dcb41f..a850850 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -87,9 +87,6 @@ importers: typescript: specifier: '>=4.7' version: 5.0.4 - undici: - specifier: ^6.23.0 - version: 6.23.0 devDependencies: '@rollup/plugin-commonjs': specifier: ^24.1.0 @@ -1803,10 +1800,6 @@ packages: resolution: {integrity: sha512-raqeBD6NQK4SkWhQzeYKd1KmIG6dllBOTt55Rmkt4HtI9mwdWtJljnrXjAFUBLTSN67HWrOIZ3EPF4kjUw80Bg==} engines: {node: '>=14.0'} - undici@6.23.0: - resolution: {integrity: sha512-VfQPToRA5FZs/qJxLIinmU59u0r7LXqoJkCzinq3ckNJp3vKEh7jTWN589YQ5+aoAC/TGRLyJLCPKcLQbM8r9g==} - engines: {node: '>=18.17'} - uri-js@4.4.1: resolution: {integrity: sha512-7rKUyy33Q1yc98pQ1DAmLtwX109F7TIfWlW1Ydo8Wl1ii1SeHieeh0HHfPeL2fMXK6z0s8ecKs9frCuLJvndBg==} @@ -3455,8 +3448,6 @@ snapshots: dependencies: '@fastify/busboy': 2.1.0 - undici@6.23.0: {} - uri-js@4.4.1: dependencies: punycode: 2.3.0