diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 26ae845..ad8ee6e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,7 +21,9 @@ jobs: strategy: fail-fast: false matrix: - node-version: [18, 20, 22] + # 22 is the floor: the real-time WebSocket client uses the global + # WebSocket, which Node ships from 22. + node-version: [22, 24] steps: - uses: actions/checkout@v4 diff --git a/README.md b/README.md index 3264a73..aa8c925 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ [![npm version](https://img.shields.io/npm/v/newsdata-nodejs-client?logo=npm&color=cb3837)](https://www.npmjs.com/package/newsdata-nodejs-client) [![npm downloads](https://img.shields.io/npm/dm/newsdata-nodejs-client?color=cb3837)](https://www.npmjs.com/package/newsdata-nodejs-client) [![CI](https://img.shields.io/github/actions/workflow/status/newsdataapi/newsdata-nodejs-client/ci.yml?branch=main&logo=github&label=CI)](https://github.com/newsdataapi/newsdata-nodejs-client/actions/workflows/ci.yml) -[![Node](https://img.shields.io/badge/node-%3E%3D18-green?logo=node.js)](https://nodejs.org) +[![Node](https://img.shields.io/badge/node-%3E%3D22-green?logo=node.js)](https://nodejs.org) [![License](https://img.shields.io/badge/license-MIT-blue)](./LICENSE) [![OpenAPI](https://img.shields.io/badge/OpenAPI-3.1-85EA2D)](https://newsdata.io/openapi.json) @@ -17,9 +17,12 @@ Official Node.js client for the [Newsdata.io](https://newsdata.io) News API. It wraps every endpoint (`latest`, `archive`, `sources`, `crypto`, `market`, `count`, `crypto/count`, `market/count`) with client-side parameter validation, automatic retries with exponential backoff, scroll/paginate helpers, and a -typed error hierarchy. +typed error hierarchy. It also covers the real-time WebSocket service end to +end with `NewsDataApiWebSocket`: register, list, and delete queries, and stream +the matching news as it is published. -Zero runtime dependencies — uses the built-in `fetch` (Node 18+). +Zero runtime dependencies — uses the built-in `fetch` and `WebSocket` +(Node 22+). ## Installation @@ -66,6 +69,9 @@ const { NewsDataApiClient } = await import('newsdata-nodejs-client'); | `countApi(params)` | `/1/count` | Aggregate counts (requires `from_date`, `to_date`) | | `cryptoCountApi(params)` | `/1/crypto/count` | Aggregate crypto counts (requires dates) | | `marketCountApi(params)` | `/1/market/count` | Aggregate market counts (requires dates) | +| `websocketRegister(params)` | `/1/websocket/register` | Register a real-time query | +| `websocketFetch()` | `/1/websocket/fetch` | List registered queries | +| `websocketDelete(id)` | `/1/websocket/delete` | Delete a registered query | Each `params` value may be a single value or an array (arrays are sent comma-separated). Parameter names are case-insensitive. See the @@ -99,6 +105,99 @@ await client.latestApi({ rawQuery: 'q=bitcoin&country=us&language=en' }); `rawQuery` is mutually exclusive with all other parameters and is validated against the endpoint's allowed keys. +## Real-time news (WebSocket) + +Register a query first — the returned `registration_id` identifies it from then on: + +```js +import { NewsDataApiClient, NewsDataApiWebSocket } from 'newsdata-nodejs-client'; + +const client = new NewsDataApiClient('YOUR_API_KEY'); +const ws = new NewsDataApiWebSocket(client); + +const { results } = await ws.websocketRegister({ q: 'bitcoin', language: 'en' }); +const registrationId = results.registration_id; +``` + +`websocketRegister` takes the familiar filter parameters (`q`, `country`, +`language`, `domain`, …) — no date or paging filters, since a registered query +matches news as it is published. Registering an identical query twice rejects +with a `NewsdataApiError` whose `statusCode` is 409; the existing id is at +`err.responseBody.results.registration_id`. `websocketFetch()` lists every +registered query and `websocketDelete(id)` removes one. + +Then stream — each response has the familiar `status` / `totalResults` / +`results` shape: + +```js +for await (const response of ws.stream(registrationId)) { + for (const article of response.results) { + console.log(article.title, '-', article.link); + } +} +``` + +Break out of the loop to stop; the connection closes either way. `ws.close()` +ends an in-flight stream from outside the loop. + +Transient drops (network errors, server restarts, abnormal closes) are +reconnected automatically with a capped exponential backoff. Pass +`reconnect: false` to stop on the first disconnect instead. A permanent +rejection — bad API key or unknown +`registration_id`, exhausted API credits, or too many simultaneous devices — throws +`NewsdataWebSocketAuthError` and is **not** retried. + +The server always accepts the handshake and then closes with code **1008** when +the connection is refused, carrying one of three reasons: `invalid credentials +or registration not found`, `api limit reached`, or `device limit reached` (more +than 5 devices on one `registration_id`). Every other close code — including +`1013` (`send timeout`, meaning the client read too slowly) — is transient and +reconnects. + +**Each delivered article consumes 1 API credit per connected device.** + +Catch it like any other client error: + +```js +import { + NewsdataWebSocketAuthError, + NewsdataWebSocketError, +} from 'newsdata-nodejs-client'; + +try { + for await (const response of ws.stream(registrationId)) { + // ... + } +} catch (err) { + if (err instanceof NewsdataWebSocketAuthError) console.error('rejected:', err.message); + else if (err instanceof NewsdataWebSocketError) console.error('stream error:', err.message); + else throw err; +} +``` + +All connection options are optional: + +```js +const ws = new NewsDataApiWebSocket(client, { + baseUrl: 'wss://ws.newsdata.io/ws/event', // staging / self-hosted / proxied + reconnect: true, // auto-reconnect on transient drops; default true + reconnectDelay: 1000, // ms before the first reconnect (doubles each retry) + reconnectDelayMax: 30000, // cap on the reconnect delay + openTimeout: 10000, // ms to wait for the opening handshake + WebSocket: undefined, // override the implementation (default: global WebSocket) +}); +``` + +> **Node 22+.** Streaming uses the global `WebSocket`, which Node ships from +> v22. On older runtimes pass your own implementation, e.g. +> `new NewsDataApiWebSocket(client, { WebSocket: require('ws') })`. +> +> Node's global `WebSocket` does not expose the handshake HTTP status. When a +> connection fails before it opens, the client probes the same URL over HTTP to +> tell a permanent rejection (401 / 403) from a transient failure. + +Runnable example: [`examples/websocket.js`](examples/websocket.js). + ## Client-side validation Before any request is sent, parameters are validated and normalized. A @@ -146,7 +245,9 @@ NewsdataError (catch-all base) │ ├── NewsdataAuthError (401 / 403) │ ├── NewsdataRateLimitError (429; .retryAfter) │ └── NewsdataServerError (5xx) -└── NewsdataNetworkError (.cause) +├── NewsdataNetworkError (.cause) +└── NewsdataWebSocketError (real-time stream) + └── NewsdataWebSocketAuthError (policy-violation close 1008) ``` ## Configuration diff --git a/examples/websocket.js b/examples/websocket.js new file mode 100644 index 0000000..893e13f --- /dev/null +++ b/examples/websocket.js @@ -0,0 +1,72 @@ +// Real-time news streaming. +// +// NEWSDATA_API_KEY= node examples/websocket.js +// +// Requires Node 22+ for the global WebSocket. +// +// Articles are matched by a registered query. If NEWSDATA_REGISTRATION_ID is +// set, that query is streamed directly; otherwise the example registers a demo +// query (q="pizza") first and prints the resulting registration_id so you can +// reuse it on the next run — or remove it later with websocketDelete(). + +import { + NewsDataApiClient, + NewsDataApiWebSocket, + NewsdataApiError, + NewsdataWebSocketAuthError, +} from '../src/index.js'; + +const apiKey = process.env.NEWSDATA_API_KEY; +if (!apiKey) { + console.error('Set NEWSDATA_API_KEY in your environment before running this example.'); + process.exit(1); +} + +const client = new NewsDataApiClient(apiKey); +const ws = new NewsDataApiWebSocket(client); + +/** + * Register q="pizza" and return its registration_id. Registering an identical + * query again answers HTTP 409 with the existing id in the response body — + * reuse it instead of failing. + */ +async function registerDemoQuery() { + try { + const { results } = await ws.websocketRegister({ q: 'pizza' }); + console.log(`registered demo query q="pizza" -> ${results.registration_id}`); + return results.registration_id; + } catch (err) { + if (err instanceof NewsdataApiError && err.statusCode === 409) { + const existing = err.responseBody?.results?.registration_id; + if (existing) { + console.log(`query already registered; reusing ${existing}`); + return existing; + } + } + throw err; + } +} + +const registrationId = process.env.NEWSDATA_REGISTRATION_ID ?? await registerDemoQuery(); + +// Stop cleanly on Ctrl-C. +process.on('SIGINT', () => { + console.log('\nstopping'); + ws.close(); +}); + +console.log(`streaming ${registrationId} — Ctrl-C to stop`); + +try { + for await (const response of ws.stream(registrationId)) { + for (const article of response.results ?? []) { + console.log(`${article.title} - ${article.link}`); + } + } +} catch (err) { + if (err instanceof NewsdataWebSocketAuthError) { + console.error(`rejected: ${err.message}`); + process.exit(1); + } + throw err; +} diff --git a/package.json b/package.json index c8a3570..2b4cc7c 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "newsdata-nodejs-client", "version": "0.1.0", - "description": "Official Node.js client (SDK) for the Newsdata.io News API — fetch real-time, historical, crypto, and stock-market news via REST with validation, retries, pagination, and typed errors.", + "description": "Official Node.js client (SDK) for the Newsdata.io News API \u2014 fetch real-time, historical, crypto, and stock-market news via REST with validation, retries, pagination, and typed errors.", "type": "module", "main": "src/index.js", "types": "types/index.d.ts", @@ -18,7 +18,7 @@ "LICENSE" ], "engines": { - "node": ">=18" + "node": ">=22" }, "scripts": { "test": "node --test" diff --git a/src/client.js b/src/client.js index a89d908..ddeaa05 100644 --- a/src/client.js +++ b/src/client.js @@ -3,6 +3,9 @@ import { BASE_URL, ENDPOINTS, + ENDPOINT_METHODS, + RESULTS_OPTIONAL, + WS_NEWS_TYPE, DEFAULT_REQUEST_TIMEOUT, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_BACKOFF, @@ -124,6 +127,67 @@ export class NewsDataApiClient { return this.#request('sources', validated); } + // ---- real-time query management --------------------------------------- + + /** + * Register a real-time WebSocket query. POST /1/websocket/register + * + * Takes the familiar filter names (`q`, `country`, `language`, `domain`, …); + * no date or paging filters apply, since a registered query matches news as + * it is published. The new query's id is at `results.registration_id` — pass + * it to `NewsDataApiWebSocket#stream`. + * + * Registering an identical query twice rejects with a `NewsdataApiError` + * whose `statusCode` is 409; the existing id is at + * `err.responseBody.results.registration_id`. + * @returns {Promise} + */ + websocketRegister(params = {}) { + const { rawQuery = null, ...rest } = params; + const validated = validateParams('websocket_register', rest, rawQuery); + validated.news_type = WS_NEWS_TYPE; + return this.#request('websocket_register', validated); + } + + /** + * List the account's registered real-time queries. GET /1/websocket/fetch + * One entry per query at `results.queries`. + * @returns {Promise} + */ + websocketFetch() { + return this.#request('websocket_fetch', {}); + } + + /** + * Delete a registered real-time query. DELETE /1/websocket/delete + * @param {string} registrationId + * @returns {Promise} + */ + websocketDelete(registrationId) { + if (typeof registrationId !== 'string' || registrationId === '') { + throw new NewsdataValidationError( + 'registrationId must be a non-empty string', + 'registration_id', + ); + } + return this.#request('websocket_delete', { registration_id: registrationId }); + } + + /** The API key, for the WebSocket handshake URL. @internal */ + get apiKeyForWebSocket() { + return this.#apiKey; + } + + /** The configured fetch, reused by the WebSocket handshake probe. @internal */ + get fetchForWebSocket() { + return this.#fetch; + } + + /** Forward a log line from the WebSocket layer. @internal */ + logForWebSocket(level, message) { + this.#log(level, message); + } + // ---- dispatch --------------------------------------------------------- /** @@ -179,16 +243,17 @@ export class NewsDataApiClient { const search = new URLSearchParams({ ...params, apikey: this.#apiKey }); const fullUrl = `${this.#endpointUrl(endpoint)}?${search.toString()}`; const logUrl = redactApiKey(fullUrl); + const method = ENDPOINT_METHODS[endpoint] ?? 'GET'; for (let attempt = 1; attempt <= this.#maxRetries; attempt += 1) { - this.#log('info', `GET ${logUrl} (attempt ${attempt}/${this.#maxRetries})`); + this.#log('info', `${method} ${logUrl} (attempt ${attempt}/${this.#maxRetries})`); let res; const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), this.#timeout); try { res = await this.#fetch(fullUrl, { - method: 'GET', + method, headers: { Accept: 'application/json' }, signal: controller.signal, }); @@ -220,7 +285,7 @@ export class NewsDataApiClient { throw new NewsdataApiError(`Non-JSON response from API (status ${status})`, status); } - if (status === 200 && this.#isSuccess(body)) { + if (status === 200 && this.#isSuccess(body, endpoint)) { if (this.#includeHeaders) { body.responseHeaders = Object.fromEntries(res.headers.entries()); } @@ -325,14 +390,11 @@ export class NewsDataApiClient { } } - #isSuccess(body) { - return ( - body - && typeof body === 'object' - && body.status === 'success' - && body.results !== null - && body.results !== undefined - ); + #isSuccess(body, endpoint) { + if (!body || typeof body !== 'object' || body.status !== 'success') return false; + // The websocket management endpoints may answer without a `results` field. + if (RESULTS_OPTIONAL.includes(endpoint)) return true; + return body.results !== null && body.results !== undefined; } #errorMessage(body, status) { diff --git a/src/constants.js b/src/constants.js index b37de6b..e1b7de0 100644 --- a/src/constants.js +++ b/src/constants.js @@ -28,8 +28,34 @@ export const ENDPOINTS = Object.freeze({ count: 'count', crypto_count: 'crypto/count', market_count: 'market/count', + websocket_register: 'websocket/register', + websocket_fetch: 'websocket/fetch', + websocket_delete: 'websocket/delete', }); +// HTTP method per endpoint; anything absent is a GET. +export const ENDPOINT_METHODS = Object.freeze({ + websocket_register: 'POST', + websocket_delete: 'DELETE', +}); + +// The websocket management endpoints answer with a success envelope that may +// carry no `results` field, so they are exempt from the results-present check +// applied to the news endpoints. +export const RESULTS_OPTIONAL = Object.freeze([ + 'websocket_register', 'websocket_fetch', 'websocket_delete', +]); + +// Real-time WebSocket defaults (NewsDataApiWebSocket). +export const WS_BASE_URL = 'wss://ws.newsdata.io/ws/event'; +// The feed a registered query matches against. +export const WS_NEWS_TYPE = 'latest'; +// Close code the server uses for a permanent rejection. +export const WS_POLICY_VIOLATION = 1008; +export const WS_RECONNECT_DELAY = 1_000; // ms before the first reconnect; doubles each retry +export const WS_RECONNECT_DELAY_MAX = 30_000; // cap on the reconnect delay +export const WS_OPEN_TIMEOUT = 10_000; // ms to wait for the opening handshake + // Endpoints that require both from_date and to_date. export const REQUIRES_DATE_RANGE = Object.freeze(['count', 'crypto_count', 'market_count']); @@ -102,6 +128,19 @@ export const FILTERS = Object.freeze({ 'market_id', 'prioritydomain', 'page', 'sentiment', 'removeduplicate', 'size', 'sort', 'tag', 'interval', 'creator', 'datatype', 'sentiment_score', ], + // Real-time query registration. No date/paging filters — a registered query + // matches news as it is published. `news_type` is set by websocketRegister, + // not by the caller. + websocket_register: [ + 'q', 'qintitle', 'qinmeta', 'country', 'excludecountry', 'category', + 'excludecategory', 'language', 'excludelanguage', 'domain', 'domainurl', + 'excludedomain', 'prioritydomain', 'timezone', 'full_content', 'image', + 'video', 'removeduplicate', 'tag', 'sentiment', 'sentiment_score', + 'region', 'organization', 'creator', 'datatype', 'excludefield', + 'news_type', + ], + websocket_fetch: [], + websocket_delete: ['registration_id'], }); // Control/meta keys accepted on endpoint methods but not sent as API params. diff --git a/src/errors.js b/src/errors.js index fa109d8..5268d3e 100644 --- a/src/errors.js +++ b/src/errors.js @@ -72,6 +72,31 @@ export class NewsdataServerError extends NewsdataApiError { } } +/** A real-time WebSocket stream failure (NewsDataApiWebSocket). */ +export class NewsdataWebSocketError extends NewsdataError { + /** + * @param {string} message + * @param {Error|null} [cause] The underlying error. + */ + constructor(message, cause = null) { + super(message); + this.name = 'NewsdataWebSocketError'; + if (cause) this.cause = cause; + } +} + +/** + * The server rejected the WebSocket connection — bad API key, missing + * WebSocket entitlement, unknown registration_id, device limit reached, or + * exhausted quota. Never retried, regardless of the `reconnect` setting. + */ +export class NewsdataWebSocketAuthError extends NewsdataWebSocketError { + constructor(message, cause = null) { + super(message, cause); + this.name = 'NewsdataWebSocketAuthError'; + } +} + /** A network-level failure (DNS, TLS, timeout, abort) prevented the request. */ export class NewsdataNetworkError extends NewsdataError { /** diff --git a/src/index.js b/src/index.js index 158fb8b..f0578a6 100644 --- a/src/index.js +++ b/src/index.js @@ -1,6 +1,7 @@ // Public entry point for the Newsdata.io Node client. export { NewsDataApiClient, redactApiKey } from './client.js'; +export { NewsDataApiWebSocket } from './websocket.js'; export { NewsdataError, NewsdataValidationError, @@ -9,6 +10,8 @@ export { NewsdataRateLimitError, NewsdataServerError, NewsdataNetworkError, + NewsdataWebSocketError, + NewsdataWebSocketAuthError, } from './errors.js'; export { validateParams } from './validator.js'; export * as constants from './constants.js'; diff --git a/src/websocket.js b/src/websocket.js new file mode 100644 index 0000000..2f89e39 --- /dev/null +++ b/src/websocket.js @@ -0,0 +1,322 @@ +// Real-time WebSocket support for NewsData.io. +// +// Uses the global `WebSocket` built into Node 22+. No dependency is required; +// pass `options.WebSocket` to supply your own implementation (e.g. `ws`) if you +// need to run somewhere it is missing. + +import { + WS_BASE_URL, + WS_POLICY_VIOLATION, + WS_RECONNECT_DELAY, + WS_RECONNECT_DELAY_MAX, + WS_OPEN_TIMEOUT, +} from './constants.js'; +import { + NewsdataError, + NewsdataValidationError, + NewsdataWebSocketError, + NewsdataWebSocketAuthError, +} from './errors.js'; +import { redactApiKey } from './client.js'; + +const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * NewsData.io real-time WebSocket service. + * + * Registers, lists, and deletes the account's real-time queries and streams + * the responses for a registered query: + * + * ```js + * const client = new NewsDataApiClient(apiKey); + * const ws = new NewsDataApiWebSocket(client); + * + * const { results } = await ws.websocketRegister({ q: 'bitcoin' }); + * + * for await (const response of ws.stream(results.registration_id)) { + * for (const article of response.results) console.log(article.title); + * } + * ``` + * + * Transient drops are reconnected automatically with a capped exponential + * backoff; pass `reconnect: false` to stop on the first disconnect. A + * permanent rejection throws `NewsdataWebSocketAuthError` and is never + * retried. + * + * Break out of the loop (or call `close()`) to stop; the connection is closed + * either way. + */ +export class NewsDataApiWebSocket { + #client; + + #baseUrl; + + #reconnect; + + #reconnectDelay; + + #reconnectDelayMax; + + #openTimeout; + + #WebSocketImpl; + + #socket = null; + + #closed = false; + + /** + * @param {import('./client.js').NewsDataApiClient} client + * Supplies the API key and performs the management HTTP calls. Not closed + * by this class. + * @param {object} [options] + * @param {string} [options.baseUrl] WebSocket endpoint. + * @param {boolean} [options.reconnect] Auto-reconnect; default true. + * @param {number} [options.reconnectDelay] ms before the first reconnect. + * @param {number} [options.reconnectDelayMax] Cap on the reconnect delay. + * @param {number} [options.openTimeout] ms to wait for the handshake. + * @param {Function} [options.WebSocket] WebSocket implementation. + */ + constructor(client, options = {}) { + if (!client) { + throw new NewsdataValidationError('client is required', 'client'); + } + this.#client = client; + this.#baseUrl = options.baseUrl ?? WS_BASE_URL; + this.#reconnect = options.reconnect ?? true; + this.#reconnectDelay = options.reconnectDelay ?? WS_RECONNECT_DELAY; + this.#reconnectDelayMax = options.reconnectDelayMax ?? WS_RECONNECT_DELAY_MAX; + this.#openTimeout = options.openTimeout ?? WS_OPEN_TIMEOUT; + this.#WebSocketImpl = options.WebSocket ?? globalThis.WebSocket; + + if (typeof this.#WebSocketImpl !== 'function') { + throw new NewsdataError( + 'No WebSocket implementation available; Node 22+ provides one globally, ' + + 'or pass options.WebSocket (e.g. the `ws` package).', + ); + } + } + + // ---- query management ------------------------------------------------- + + /** Register a real-time query. See NewsDataApiClient#websocketRegister. */ + websocketRegister(params = {}) { + return this.#client.websocketRegister(params); + } + + /** List registered queries. See NewsDataApiClient#websocketFetch. */ + websocketFetch() { + return this.#client.websocketFetch(); + } + + /** Delete a registered query. See NewsDataApiClient#websocketDelete. */ + websocketDelete(registrationId) { + return this.#client.websocketDelete(registrationId); + } + + // ---- streaming -------------------------------------------------------- + + #url(registrationId) { + const search = new URLSearchParams({ + apikey: this.#client.apiKeyForWebSocket, + registration_id: registrationId, + }); + return `${this.#baseUrl}?${search.toString()}`; + } + + #nextDelay(delay) { + return Math.min(delay * 2, this.#reconnectDelayMax); + } + + /** + * Connect and yield each response for `registrationId` as it arrives. + * Responses have the familiar status / totalResults / results shape. + * + * @param {string} registrationId + * @returns {AsyncGenerator} + */ + async* stream(registrationId) { + if (typeof registrationId !== 'string' || registrationId === '') { + throw new NewsdataValidationError( + 'registrationId must be a non-empty string', + 'registration_id', + ); + } + const url = this.#url(registrationId); + const logUrl = redactApiKey(url); + let delay = this.#reconnectDelay; + this.#closed = false; + + try { + while (!this.#closed) { + const session = this.#connect(url, logUrl); + + try { + for await (const message of session) { + delay = this.#reconnectDelay; // reset after a successful connect + let response; + try { + response = JSON.parse(message); + } catch { + continue; // skip malformed frames + } + if (response && typeof response === 'object' && !Array.isArray(response)) { + yield response; + } + } + } catch (err) { + if (this.#closed) return; + const permanent = this.#permanentAuthError(err); + if (permanent) throw permanent; + if (!this.#reconnect) throw toTransientError(err); + this.#client.logForWebSocket( + 'warn', + `connection to ${logUrl} failed (${err.message}); reconnecting in ${delay}ms`, + ); + } + + if (this.#closed) return; + // A clean close with reconnect disabled ends the stream. + if (!this.#reconnect) return; + await sleep(delay); + delay = this.#nextDelay(delay); + } + } finally { + this.close(); + } + } + + /** + * Bridge one WebSocket connection's events into an async iterable of raw + * message payloads. Throws a `WsClosed` when the socket drops. + */ + #connect(url, logUrl) { + const socket = new this.#WebSocketImpl(url); + this.#socket = socket; + + /** @type {string[]} */ + const queue = []; + /** @type {{resolve: Function, reject: Function}[]} */ + const waiters = []; + let failure = null; + let done = false; + + const settle = () => { + while (waiters.length) { + const waiter = waiters.shift(); + if (queue.length) waiter.resolve({ value: queue.shift(), done: false }); + else if (failure) waiter.reject(failure); + else if (done) waiter.resolve({ value: undefined, done: true }); + else { + waiters.unshift(waiter); + return; + } + } + }; + + let openTimer = null; + if (this.#openTimeout > 0) { + openTimer = setTimeout(() => { + failure = new WsClosed('handshake timed out', null, false); + try { socket.close(); } catch { /* already closing */ } + settle(); + }, this.#openTimeout); + } + + socket.addEventListener('open', () => { + if (openTimer) clearTimeout(openTimer); + this.#client.logForWebSocket('info', `connected to ${logUrl}`); + }); + + socket.addEventListener('message', (event) => { + const { data } = event; + queue.push(typeof data === 'string' ? data : String(data)); + settle(); + }); + + socket.addEventListener('error', () => { + // The close event that follows carries the code; record nothing here so + // the close handler classifies the failure. + }); + + socket.addEventListener('close', (event) => { + if (openTimer) clearTimeout(openTimer); + const code = event?.code ?? null; + const reason = event?.reason || ''; + if (code === 1000) { + done = true; // normal closure + } else { + failure = new WsClosed(reason || `connection closed (${code})`, code, true); + } + settle(); + }); + + return { + [Symbol.asyncIterator]() { + return { + next() { + if (queue.length) { + return Promise.resolve({ value: queue.shift(), done: false }); + } + if (failure) return Promise.reject(failure); + if (done) return Promise.resolve({ value: undefined, done: true }); + return new Promise((resolve, reject) => { + waiters.push({ resolve, reject }); + }); + }, + return() { + try { socket.close(); } catch { /* already closing */ } + return Promise.resolve({ value: undefined, done: true }); + }, + }; + }, + }; + } + + /** + * Decide whether a failure is a permanent rejection. + * + * The server always accepts the handshake and then closes with code 1008 + * on a permanent failure — bad apikey or unknown registration_id + * ("invalid credentials or registration not found"), exhausted credits + * ("api limit reached"), or too many simultaneous devices for the same + * registration_id ("device limit reached"). Every other close code, + * including 1013 ("send timeout", the client read too slowly), is + * transient and reconnects. + * + * @returns {NewsdataWebSocketAuthError|null} + */ + #permanentAuthError(err) { + if (err instanceof WsClosed && err.code === WS_POLICY_VIOLATION) { + return new NewsdataWebSocketAuthError(err.message || 'connection rejected', err); + } + return null; + } + + /** Close the active connection, ending any in-flight `stream()`. */ + close() { + this.#closed = true; + if (this.#socket) { + try { this.#socket.close(); } catch { /* already closing */ } + this.#socket = null; + } + } +} + +/** Internal marker for a dropped connection, carrying the close code. */ +class WsClosed extends Error { + constructor(message, code, wasOpen) { + super(message); + this.name = 'WsClosed'; + this.code = code; + this.wasOpen = wasOpen; + } +} + +/** Wrap a transient failure, used only when reconnect is disabled. */ +function toTransientError(err) { + if (err instanceof WsClosed) { + return new NewsdataWebSocketError(err.code === null ? err.message : 'connection closed', err); + } + return new NewsdataWebSocketError(`connection error: ${err.message}`, err); +} diff --git a/test/websocket.test.js b/test/websocket.test.js new file mode 100644 index 0000000..9dc6ccb --- /dev/null +++ b/test/websocket.test.js @@ -0,0 +1,331 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { NewsDataApiClient } from '../src/client.js'; +import { NewsDataApiWebSocket } from '../src/websocket.js'; +import { + NewsdataValidationError, + NewsdataWebSocketAuthError, + NewsdataWebSocketError, +} from '../src/errors.js'; + +function mockResponse(status, body) { + return { + status, + headers: new Headers(), + text: async () => (typeof body === 'string' ? body : JSON.stringify(body)), + }; +} + +function stubFetch(responses) { + const calls = []; + const queue = [...responses]; + const fn = async (url, init = {}) => { + calls.push({ url, method: init.method ?? 'GET' }); + const next = queue.shift(); + if (typeof next === 'function') return next(); + return next ?? mockResponse(200, { status: 'success', results: {} }); + }; + fn.calls = calls; + return fn; +} + +/** + * A scriptable stand-in for the global WebSocket. `script` runs with the + * socket instance once listeners are attached, and drives the events. + */ +function fakeWebSocketFactory(script) { + const instances = []; + class FakeWebSocket { + constructor(url) { + this.url = url; + this.listeners = new Map(); + this.closed = false; + instances.push(this); + // Let the caller attach listeners before anything fires. + queueMicrotask(() => script(this, instances.length)); + } + + addEventListener(type, fn) { + if (!this.listeners.has(type)) this.listeners.set(type, []); + this.listeners.get(type).push(fn); + } + + emit(type, event) { + for (const fn of this.listeners.get(type) ?? []) fn(event); + } + + open() { this.emit('open', {}); } + + message(data) { + this.emit('message', { data: typeof data === 'string' ? data : JSON.stringify(data) }); + } + + drop(code = 1006, reason = '') { this.emit('close', { code, reason }); } + + close() { + if (this.closed) return; + this.closed = true; + this.emit('close', { code: 1000, reason: '' }); + } + } + FakeWebSocket.instances = instances; + return FakeWebSocket; +} + +const article = (id, title) => JSON.stringify({ + status: 'success', totalResults: 1, results: [{ article_id: id, title }], +}); + +function wsClient(fetchStub = stubFetch([])) { + return new NewsDataApiClient('key', { fetch: fetchStub }); +} + +test('stream yields each response as it arrives', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.message(article('a1', 'one')); + socket.message(article('a2', 'two')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + const titles = []; + for await (const response of ws.stream('reg-1')) { + titles.push(response.results[0].title); + if (titles.length === 2) break; + } + assert.deepEqual(titles, ['one', 'two']); +}); + +test('stream sends apikey and registration_id in the query', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.message(article('a1', 'one')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-42')) break; + + const { url } = FakeWebSocket.instances[0]; + assert.match(url, /apikey=key/); + assert.match(url, /registration_id=reg-42/); +}); + +test('stream skips malformed frames', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.message('not json at all'); + socket.message(article('a1', 'one')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + const seen = []; + for await (const response of ws.stream('reg-1')) { + seen.push(response.results[0].title); + break; + } + assert.deepEqual(seen, ['one'], 'the malformed frame should be skipped, not yielded'); +}); + +test('close code 1008 raises a permanent auth error and does not reconnect', async () => { + let connections = 0; + const FakeWebSocket = fakeWebSocketFactory((socket, n) => { + connections = n; + socket.drop(1008, 'quota exhausted'); + }); + // reconnect stays ON to prove a permanent rejection is not retried. + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + await assert.rejects( + async () => { + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-1')) { /* unreachable */ } + }, + (err) => { + assert.ok(err instanceof NewsdataWebSocketAuthError, `got ${err.name}`); + assert.match(err.message, /quota exhausted/); + return true; + }, + ); + assert.equal(connections, 1, 'a permanent rejection must not retry'); +}); + +// The server always accepts the handshake, then closes with 1008 carrying the +// reason. These are the three documented permanent rejections. +for (const reason of [ + 'invalid credentials or registration not found', + 'api limit reached', + 'device limit reached', +]) { + test(`close 1008 "${reason}" is permanent`, async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.drop(1008, reason); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + await assert.rejects( + async () => { + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-1')) { /* unreachable */ } + }, + (err) => { + assert.ok(err instanceof NewsdataWebSocketAuthError, `got ${err.name}`); + assert.match(err.message, new RegExp(reason)); + return true; + }, + ); + }); +} + +// 1013 ("send timeout" — the client read too slowly) is transient. +test('close 1013 is transient and reconnects', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket, n) => { + if (n === 1) { + socket.open(); + socket.drop(1013, 'send timeout'); + return; + } + socket.open(); + socket.message(article('a1', 'after-reconnect')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { + WebSocket: FakeWebSocket, + reconnectDelay: 1, + reconnectDelayMax: 2, + }); + + const titles = []; + for await (const response of ws.stream('reg-1')) { + titles.push(response.results[0].title); + break; + } + assert.deepEqual(titles, ['after-reconnect']); +}); + +test('a transient drop stops with a websocket error when reconnect is disabled', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.drop(1006); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { + WebSocket: FakeWebSocket, + reconnect: false, + }); + + await assert.rejects( + async () => { + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-1')) { /* unreachable */ } + }, + (err) => { + assert.ok(err instanceof NewsdataWebSocketError, `got ${err.name}`); + assert.ok(!(err instanceof NewsdataWebSocketAuthError), 'should not be an auth error'); + return true; + }, + ); +}); + +test('a transient drop reconnects when reconnect is enabled', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket, n) => { + if (n === 1) { + socket.open(); + socket.drop(1006); // transient + return; + } + socket.open(); + socket.message(article('a1', 'after-reconnect')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { + WebSocket: FakeWebSocket, + reconnectDelay: 1, + reconnectDelayMax: 2, + }); + + const titles = []; + for await (const response of ws.stream('reg-1')) { + titles.push(response.results[0].title); + break; + } + assert.deepEqual(titles, ['after-reconnect']); + assert.ok(FakeWebSocket.instances.length >= 2, 'should have reconnected'); +}); + +test('breaking out of the loop closes the socket', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.message(article('a1', 'one')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-1')) break; + + assert.equal(FakeWebSocket.instances[0].closed, true); +}); + +test('stream rejects an empty registration id', async () => { + const FakeWebSocket = fakeWebSocketFactory(() => {}); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + await assert.rejects( + async () => { + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('')) { /* unreachable */ } + }, + NewsdataValidationError, + ); +}); + +// ---- query management --------------------------------------------------- + +test('websocketRegister POSTs and injects news_type=latest', async () => { + const fetchStub = stubFetch([ + mockResponse(200, { status: 'success', results: { registration_id: 'reg-9' } }), + ]); + const client = wsClient(fetchStub); + const res = await client.websocketRegister({ q: 'bitcoin' }); + + assert.equal(res.results.registration_id, 'reg-9'); + assert.equal(fetchStub.calls[0].method, 'POST'); + assert.match(fetchStub.calls[0].url, /news_type=latest/); + assert.match(fetchStub.calls[0].url, /q=bitcoin/); + assert.match(fetchStub.calls[0].url, /websocket\/register/); +}); + +test('websocketFetch GETs the fetch endpoint', async () => { + const fetchStub = stubFetch([ + mockResponse(200, { status: 'success', results: { queries: [] } }), + ]); + await wsClient(fetchStub).websocketFetch(); + assert.equal(fetchStub.calls[0].method, 'GET'); + assert.match(fetchStub.calls[0].url, /websocket\/fetch/); +}); + +test('websocketDelete uses DELETE and carries registration_id', async () => { + const fetchStub = stubFetch([ + mockResponse(200, { status: 'success', results: { deleted: true } }), + ]); + await wsClient(fetchStub).websocketDelete('reg-9'); + assert.equal(fetchStub.calls[0].method, 'DELETE'); + assert.match(fetchStub.calls[0].url, /registration_id=reg-9/); +}); + +test('websocketDelete rejects an empty id', () => { + assert.throws(() => wsClient().websocketDelete(''), NewsdataValidationError); +}); + +test('a resultless success envelope still succeeds on the websocket endpoints', async () => { + const fetchStub = stubFetch([mockResponse(200, { status: 'success' })]); + const res = await wsClient(fetchStub).websocketDelete('reg-9'); + assert.equal(res.status, 'success'); +}); + +test('the WebSocket class delegates management calls to the client', async () => { + const fetchStub = stubFetch([ + mockResponse(200, { status: 'success', results: { registration_id: 'reg-7' } }), + ]); + const FakeWebSocket = fakeWebSocketFactory(() => {}); + const ws = new NewsDataApiWebSocket(wsClient(fetchStub), { WebSocket: FakeWebSocket }); + const res = await ws.websocketRegister({ q: 'x' }); + assert.equal(res.results.registration_id, 'reg-7'); +}); diff --git a/types/index.d.ts b/types/index.d.ts index ec96a02..aeda949 100644 --- a/types/index.d.ts +++ b/types/index.d.ts @@ -64,6 +64,49 @@ export class NewsDataApiClient { cryptoCountApi(params?: EndpointParams): EndpointResult; marketCountApi(params?: EndpointParams): EndpointResult; sourcesApi(params?: EndpointParams): Promise; + + /** Register a real-time WebSocket query. POST /1/websocket/register */ + websocketRegister(params?: EndpointParams): Promise; + /** List the account's registered real-time queries. GET /1/websocket/fetch */ + websocketFetch(): Promise; + /** Delete a registered real-time query. DELETE /1/websocket/delete */ + websocketDelete(registrationId: string): Promise; +} + +export interface WebSocketOptions { + /** WebSocket endpoint; defaults to wss://ws.newsdata.io/ws/event. */ + baseUrl?: string; + /** Reconnect automatically on transient drops. Default true. */ + reconnect?: boolean; + /** Milliseconds before the first reconnect; doubles after each failure. */ + reconnectDelay?: number; + /** Upper bound on the reconnect delay, in milliseconds. */ + reconnectDelayMax?: number; + /** Milliseconds to wait for the opening handshake. */ + openTimeout?: number; + /** WebSocket implementation; defaults to the global one (Node 22+). */ + WebSocket?: new (url: string) => unknown; +} + +/** + * NewsData.io real-time WebSocket service: registers, lists, and deletes the + * account's real-time queries, and streams the responses for one of them. + */ +export class NewsDataApiWebSocket { + constructor(client: NewsDataApiClient, options?: WebSocketOptions); + + /** Register a real-time query. */ + websocketRegister(params?: EndpointParams): Promise; + /** List the account's registered real-time queries. */ + websocketFetch(): Promise; + /** Delete a registered real-time query. */ + websocketDelete(registrationId: string): Promise; + + /** Yield each response for `registrationId` as it arrives. */ + stream(registrationId: string): AsyncGenerator; + + /** Close the active connection, ending any in-flight `stream()`. */ + close(): void; } export function redactApiKey(url: string): string; @@ -89,5 +132,9 @@ export class NewsdataServerError extends NewsdataApiError {} export class NewsdataNetworkError extends NewsdataError { cause?: Error; } +export class NewsdataWebSocketError extends NewsdataError { + cause?: Error; +} +export class NewsdataWebSocketAuthError extends NewsdataWebSocketError {} export as namespace newsdata;