diff --git a/README.md b/README.md index bcc8991..f743168 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,10 @@ SDK to access [Zenrows](https://www.zenrows.com/) API directly from Node.js. Zen - [POST Requests](#post-requests) - [Extract](#extract) - [Batch](#batch) + - [Results and retention](#results-and-retention) - [Failed runs and extensible values](#failed-runs-and-extensible-values) + - [Errors and retries](#errors-and-retries) + - [Webhooks](#webhooks) - [Extract in a batch](#extract-in-a-batch) - [Crawl](#crawl) - [Concurrency](#concurrency) @@ -201,56 +204,174 @@ A `422` with `code: "no_data_extracted"` means Extract knows this kind of page b ### Batch -The [Batch API](https://docs.zenrows.com) runs many URLs asynchronously as a job, and is reachable via `client.batch`. A job is either **open** (it keeps accepting tasks via `addTasks()` until you `closeJob()` it, useful when you're streaming URLs in over time) or a normal one-shot batch where every task is known upfront. +The [Batch API](https://docs.zenrows.com/batch/introduction) runs many URLs asynchronously as a job, and is reachable via `client.batch`. A job is either a one-shot batch where every task is known upfront (`submitRegular()`), or **open** (`submitOpen()`): it keeps accepting tasks via `addTasks()` until you close it, which is useful when you're streaming URLs in over time. Batch costs the same credits as the regular API. -```javascript -const { ZenRows } = require("zenrows"); +`submitRegular()`, `submitOpen()` and `submitScheduled()` return a `JobRef`. Work with the job through it: `job.addTasks()`, `job.close()`, `job.rerun()`, `job.retryFailed()`, `job.delete()`, and the current run under `job.run` (`wait()`, `results()`, `downloadToDir()`, `stop()`, `pause()`, `resume()`). -const apiKey = "YOUR-API-KEY"; +```ts +import { ZenRows } from "zenrows"; -(async () => { - const client = new ZenRows(apiKey); +const client = new ZenRows("YOUR-API-KEY"); - // One-shot batch: every task known upfront. - const job = await client.batch.submitJob({ - tasks: [{ url: "https://example.com/1" }, { url: "https://example.com/2" }], - }); +// One-shot batch: every task known upfront. Strings and { url, ... } objects both work. +const job = await client.batch.submitRegular( + [ + "https://example.com/1", + { url: "https://example.com/2", external_id: "page-2" }, + // A POST task: `method` plus a `body` (a JSON value or a string, up to 16 KiB). + { url: "https://example.com/search", method: "POST", body: { q: "shoes" } }, + ], + undefined, + // A retried submit with the same key returns the first job instead of creating another. + { idempotencyKey: "catalog-2026-10-16" }, +); - // Streaming batch: keep the job open, add tasks as they arrive, close when done. - const streamingJob = await client.batch.submitJob({ status: "open" }); - await client.batch.addTasks(streamingJob.job_id, [{ url: "https://example.com/3" }]); - // Or close it as part of the last addTasks() call instead of a separate request: - // await client.batch.addTasks(streamingJob.job_id, [...], { lastBatch: true }); - await client.batch.closeJob(streamingJob.job_id); +// Wait for the run to end, then read its results. +const run = await job.run.wait(); +if (run.status === "failed") { + // The run stopped for an account reason, such as running out of credits. + console.log(run.data.failure_reason, run.data.failure_detail); +} +for await (const row of job.run.results({ status: "successful" })) { + console.log(row.task_id, row.url, row.status, row.result_url); +} + +// Or save every successful result to disk. Rows with no body to fetch (an `expired` result, +// or a `failed` task) are skipped and passed to `onSkip`. +const saved = await job.run.downloadToDir("./results", { + onSkip: (row) => console.log("skipped", row.taskId, row.status), +}); +console.log(`saved ${saved} files`); + +// Retry only the tasks that failed, inheriting the rest from the previous run. With no +// failed task there is nothing to retry: the API answers 409 `no_matching_tasks`. +if (run.stats.failed > 0) { + const retry = await job.retryFailed({ waitForIngest: true }); + await retry.wait(); +} + +// Streaming batch: keep the job open, add tasks as they arrive, close with the last call. +const streaming = await client.batch.submitOpen(); +// With an idempotencyKey, a retried call adds its tasks once, so the SDK retries it safely. +await streaming.addTasks([{ url: "https://example.com/3" }], { idempotencyKey: "chunk-1" }); +await streaming.addTasks([{ url: "https://example.com/4" }], { + idempotencyKey: "chunk-2", + lastBatch: true, +}); + +// Pick up an existing job later by its id. +const sameJob = await client.batch.getJob(job.jobId); +console.log(sameJob.status, sameJob.data.latest_run?.stats); +``` - // Poll for progress and page through results once it's done. - const status = await client.batch.getJob(job.job_id); - const { results, next_cursor } = await client.batch.getResults(job.job_id); +The client also wraps the rest of the API: - // Retry only the tasks that failed, inheriting the rest from the previous run. - await client.batch.rerun(job.job_id, { status: "failed" }); -})(); -``` +- Jobs and runs: `getJob()`, `listJobs()` / `iterJobs()`, `job.runs()`, `getRun()`, `run(jobId, runId).delete()`, `waitForRun()`. +- Results: `getResults()` / `iterResults()`, `downloadToMemory()`, `downloadTaskToFile()`, `downloadTaskToMemory()`, `getTaskHistory()`, `getTaskContent()` (the content as text) and `getTaskContentBytes()` (a `Buffer`). Text decoding corrupts binary bodies, so read a `pdf` or other binary result with `getTaskContentBytes()`. +- Scheduling: `submitScheduled()` with the `At`, `Rate` and `Calendar` builders, and `job.schedule.pause()` / `resume()` / `update()`. +- Webhooks: `job.getWebhook()`, `job.setWebhook(url, signature)`, `putJobWebhook()`, `job.deleteWebhook()`, `testWebhook()`, and `verifyWebhookSignature()` for your receiver (see [Webhooks](#webhooks)). +- HMAC signing keys: `listHmacKeys()`, `rotateHmacKey()`, `finalizeHmacKey()`, `cancelHmacRotation()`. +- CSV task uploads: `uploadCsv()`, then `submitRegular(undefined, fileInputId)`. +- Results exports: `job.run.startExport()`, `waitForExport()`, `downloadAllResults()`. +- Cost: `estimateCost(tasks, zenrowsParams)` prices each task on the job params overlaid with its own `zenrows_params`, locally, with no API call. A flag counts as on only when it is `true` or `"true"`, the values the API accepts. + +A submit of 10,000 tasks or more, and every rerun, returns while the API is still writing the run's task rows (`ingest_status: "pending"`). Pass `waitForIngest: true` to `submitRegular()`, `submitOpen()`, `rerun()` or `retryFailed()` to resolve only once the rows are written; until then `results()` can be partial. A scheduled submit (`submitScheduled()`) creates no run: it stores the task template, and the first run starts when the schedule fires. -`client.batch` also exposes `listJobs()`, `deleteJob()`, `stopRun()`, `rerun()`, `listRuns()`, `getRun()`, `deleteRun()`, and `getTaskContent()` (returns the scraped page's raw content as a string, not JSON — the endpoint can return HTML or plain text depending on what the target page served). Scheduling, webhook config, HMAC key rotation, CSV task uploads, and results exports aren't wrapped yet — call the [Batch API](https://docs.zenrows.com) directly for those. +#### Results and retention + +A result's `result_url` is valid for 2 hours. To get a fresh one, list the results again or read the content with `getTaskContent()`. Results are kept for 14 days by default. After that a successful task comes back with `status: "expired"` and no `result_url`, and `getTaskContent()` throws a `ZenRowsBatchError` with status 410 and code `result_expired`. A `status: "successful"` filter still matches expired rows, which is why the bulk downloads skip them. #### Failed runs and extensible values -A run the server auto-fails carries `failure_reason` and a human-readable `failure_detail`. Starting, rerunning or resuming a job whose API key hit one of its credit caps throws a `ZenRowsBatchError` with `status` 402 and `code` `"api_key_cap_reached"`; `error.problem.detail` names the cap and when it resets (UTC). +A run that stops for an account reason ends `failed`, with `failure_reason` (`insufficient_credits`, `subscription_inactive` or `api_key_cap_reached`) and a human-readable `failure_detail`. `job.run.wait()` resolves with that run; pass `failureStatuses: new Set(["failed"])` to make it throw instead. A paused run makes no progress until it is resumed, so the wait stops at once and throws a `RunPausedError` whose `run` is the paused run; pass `waitWhilePaused: true` to keep polling, for example when another process will resume it. A wait that runs out of time throws a `WaiterTimeoutError`. + +```ts +import { ZenRowsBatchClient } from "zenrows"; -```js -const run = await client.batch.getRun(jobId, runId); +const batch = new ZenRowsBatchClient("YOUR-API-KEY"); +const run = await batch.getRun("YOUR-JOB-ID", "YOUR-RUN-ID"); if (run.data.failure_reason === "api_key_cap_reached") console.log(run.data.failure_detail); ``` -Enum-valued fields on responses (run, job, task and export `status`, `ingest_status`, `failure_reason`) are typed `Extensible<...>`: the known values plus any `string`, because the server may add values. The exported named types (`RunStatus`, `JobStatus`, `TaskStatus`, `ExportStatus`, `IngestStatus`, `RunFailureReason`) stay closed, so they still work as request input. Give switches on these fields a default branch; an exhaustive `switch` that assigns the leftover to `never` no longer compiles. +Enum-valued fields on responses (run, job, task and export `status`, `ingest_status`, `failure_reason`, run `trigger`) are typed `Extensible<...>`: the known values plus any `string`, because the server may add values. The exported named types (`RunStatus`, `JobStatus`, `TaskStatus`, `ExportStatus`, `IngestStatus`, `RunFailureReason`, `RunTrigger`) stay closed, so they still work as request input. Give switches on these fields a default branch; an exhaustive `switch` that assigns the leftover to `never` no longer compiles. + +#### Errors and retries + +A non-2xx answer throws a `ZenRowsBatchError` with `status`, `code`, `detail` and `retryAfter` (seconds, when the API sent `Retry-After`). `BatchErrorCode` names every code, so you can branch without bare strings: + +```ts +import { BatchErrorCode, ZenRowsBatchClient, ZenRowsBatchError } from "zenrows"; + +const batch = new ZenRowsBatchClient("YOUR-API-KEY"); +try { + await batch.submitRegular(["https://example.com"]); +} catch (error) { + if (!(error instanceof ZenRowsBatchError)) throw error; + switch (error.code) { + case BatchErrorCode.PaymentRequired: // 402: the account is out of credits + case BatchErrorCode.SubscriptionInactive: // 401: the subscription is not active + console.log("Check Billing in your dashboard:", error.detail); + break; + case BatchErrorCode.ApiKeyCapReached: // 402: `detail` names the cap and when it resets (UTC) + console.log(error.detail); + break; + case BatchErrorCode.QuotaExceeded: // 429: an account limit, see below + console.log("Wait for a job to finish, or delete one:", error.detail); + break; + default: + throw error; + } +} +``` + +`quota_exceeded` means one of the account limits: 3 active jobs (shared with Crawl), 5 scheduled jobs, or 100,000 tasks in an open job. It clears only when a job finishes or is deleted, so the client never retries it. Other transient failures (network errors, 502, 503, 504, and any other 429) are retried up to `retries` times (default 3) with backoff, on reads and on the POSTs that carry an `idempotencyKey`: submit, rerun and `addTasks()`. The API keeps a key for 24 hours: the same key with the same body returns the original answer, and the same key with a different body is a 409 `idempotency_key_conflict`. + +#### Webhooks + +A job with a webhook gets one event when a run completes (`run.completed`) or fails on its own (`run.failed`, for example when the account runs out of credits). The event carries the run's stats, see the `WebhookEvent` type. Stopping a run yourself sends no event. A delivery that fails or takes longer than 10 seconds is retried with backoff for up to 3 days, so the same event can arrive more than once: deduplicate on the `X-ZenRows-Event-Id` header. + +With `signature: true`, each delivery has an `X-Signature` header, `t=,v1=,kid=`. `verifyWebhookSignature()` checks it against the secret `rotateHmacKey()` returned (base64, shown once), and rejects deliveries more than 5 minutes old by default. During a key rotation, pass both key ids in `secrets`; the header's `kid` picks the one to check. Pass the raw body, before any JSON parsing: + +```ts +import { createServer } from "node:http"; +import { + WEBHOOK_EVENT_ID_HEADER, + WEBHOOK_SIGNATURE_HEADER, + type WebhookEvent, + verifyWebhookSignature, +} from "zenrows"; + +// The `kid` and base64 `secret` that rotateHmacKey() returned. Load them from your secret store. +const secrets = { "YOUR-KEY-ID": "YOUR-BASE64-SECRET" }; + +createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (chunk: Buffer) => chunks.push(chunk)); + req.on("end", () => { + const body = Buffer.concat(chunks); + const signature = req.headers[WEBHOOK_SIGNATURE_HEADER.toLowerCase()]; + if (typeof signature !== "string" || !verifyWebhookSignature(body, signature, secrets)) { + res.writeHead(401).end(); + return; + } + const event = JSON.parse(body.toString("utf8")) as WebhookEvent; + console.log(req.headers[WEBHOOK_EVENT_ID_HEADER.toLowerCase()], event.event_type, event.stats); + res.writeHead(204).end(); + }); +}).listen(3000); +``` + +Changing a webhook with `putJobWebhook(jobId, { url, signature })`: pass `signature` every time. If you leave it out, the SDK reads the job's current setting and sends it back, so changing only the URL keeps signing as it was; that is a read then a write, so a change someone else makes in between is overwritten. #### Extract in a batch -Set `extract` in the batch params to run tasks through Extract — structured data instead of raw HTML. It works job-wide or per task, and per-task values win on collision. +Set `extract` in the batch params to run tasks through Extract: structured data instead of raw HTML. It works job-wide or per task, and per-task values win on collision. `extract_fields` is accepted the same way. -```js -const job = await client.batch.submitRegular( +```ts +import { ZenRowsBatchClient } from "zenrows"; + +const batch = new ZenRowsBatchClient("YOUR-API-KEY"); +const job = await batch.submitRegular( [ // Inherits the job-level params below. { url: "https://example.com/products", external_id: "p1" }, @@ -260,16 +381,22 @@ const job = await client.batch.submitRegular( undefined, { zenrowsParams: { extract: "auto" } }, ); +console.log(job.jobId); ``` -An Extract task's result carries two keys — `html` (the raw page) and `parsed` (the structured data). It costs the same as a regular task, so `estimateCost` prices it correctly. `extract_fields` is not supported in Batch yet. +An Extract task's result carries two keys: `html` (the raw page) and `parsed` (the structured data). `estimateCost` prices it like a regular task. -The batch client (`ZenRowsBatchClient`) also works standalone, without a `ZenRows` instance — matching the Go and Python SDKs' batch clients: +The batch client (`ZenRowsBatchClient`) also works standalone, without a `ZenRows` instance, matching the Go and Python SDKs' batch clients: -```javascript -const { ZenRowsBatchClient } = require("zenrows"); +```ts +import { ZenRowsBatchClient } from "zenrows"; -const batch = new ZenRowsBatchClient(apiKey, { baseURL: "https://async.api.zenrows.com/v1" }); // baseURL is optional +// Both settings are optional. `timeout` is seconds per API request (default 30). +const batch = new ZenRowsBatchClient("YOUR-API-KEY", { + baseURL: "https://async.api.zenrows.com/v1", + timeout: 30, +}); +console.log(batch.apiKey.length > 0); ``` ### Crawl diff --git a/src/batch/client.ts b/src/batch/client.ts index 0d70c45..5da9304 100644 --- a/src/batch/client.ts +++ b/src/batch/client.ts @@ -39,13 +39,15 @@ import type { Run, StartExportResponse, TaskHistoryResponse, + TaskMethod, TaskResult, TestWebhookResponse, WebhookConfig, } from "./types.js"; -import { pollUntil } from "./waiters.js"; +import { RunPausedError, pollUntil } from "./waiters.js"; const DEFAULT_BATCH_API_URL = "https://async.api.zenrows.com/v1"; +const DEFAULT_TIMEOUT_SECONDS = 30; // Run states that never transition again: the default waiter target. `failed` is an // account-level auto-fail (e.g. `api_key_cap_reached`, `insufficient_credits`). @@ -55,8 +57,17 @@ const TERMINAL_EXPORT_STATUSES = new Set(["completed", "failed"]); export interface BatchClientConfig { /** Override the Batch API base URL. Matches the Go/Python batch clients' base-URL override. */ baseURL?: string; - /** Retries for transient failures (429/502/503/504 + network errors) on idempotent requests. Default 3. */ + /** + * Retries for transient failures (429/502/503/504 + network errors) on idempotent requests. + * Default 3. A 429 `quota_exceeded` (an account limit) is never retried. + */ retries?: number; + /** + * Seconds each API request may take. Default 30. Waiters (`wait`, `waitForRun`) poll with + * many short requests, so this does not limit how long a wait lasts, and result downloads + * from `result_url` are not limited by it either. + */ + timeout?: number; } export interface BatchTask { @@ -64,10 +75,18 @@ export interface BatchTask { external_id?: string; metadata?: Record; zenrows_params?: Record; + /** HTTP method for the target fetch. Default `GET`. */ + method?: TaskMethod; + /** Request body for a `POST` task: a JSON value or a string, up to 16 KiB. */ + body?: unknown; } export interface BatchWebhookConfig { url: string; + /** + * Sign deliveries with the account's HMAC key. On `putJobWebhook`, pass it every time; + * see that method for what happens when it is left out. + */ signature?: boolean; } @@ -93,6 +112,11 @@ export interface SubmitTypedOptions { metadata?: Record; webhook?: BatchWebhookConfig; idempotencyKey?: string; + /** + * Block until the run's task rows are written for a large (202) submission. No effect on + * `submitScheduled`: a scheduled submit creates no run, the first one starts when the + * schedule fires. + */ waitForIngest?: boolean; } @@ -106,6 +130,10 @@ export interface ListJobsOptions { export interface GetResultsOptions { runId?: string; cursor?: string; + /** + * @deprecated Ignored and not sent: the results endpoint returns pages of a fixed size + * (about 1,000 rows). Follow `next_cursor`, or use `iterResults`. + */ limit?: number; status?: string; } @@ -126,6 +154,12 @@ export interface WaitForRunOptions { timeout?: number; pollInterval?: number; maxPollInterval?: number; + /** + * A paused run never progresses until someone resumes it, so by default the wait stops + * at once with a `RunPausedError` carrying the run. Set this to keep polling while + * paused, for example when another process will resume it. + */ + waitWhilePaused?: boolean; } type TaskInput = @@ -136,14 +170,13 @@ type TaskInput = metadata?: Record; /** Per-task scrape params, merged over the job-level `zenrowsParams` (task wins). */ zenrows_params?: Record; + /** HTTP method for the target fetch. Default `GET`. */ + method?: TaskMethod; + /** Request body for a `POST` task: a JSON value or a string, up to 16 KiB. */ + body?: unknown; }; -function coerceUrl(item: TaskInput): { - url: string; - external_id?: string; - metadata?: Record; - zenrows_params?: Record; -} { +function coerceUrl(item: TaskInput): BatchTask { return typeof item === "string" ? { url: item } : item; } @@ -167,6 +200,8 @@ export class ZenRowsBatchClient { config.baseURL ?? DEFAULT_BATCH_API_URL, apiKey, config.retries, + undefined, + (config.timeout ?? DEFAULT_TIMEOUT_SECONDS) * 1000, ); } @@ -325,7 +360,7 @@ export class ZenRowsBatchClient { // ===== results / content ===== async getResults(jobId: string, options: GetResultsOptions = {}): Promise { - const { runId, ...query } = options; + const { runId, limit: _ignored, ...query } = options; const path = runId ? `/jobs/${jobId}/runs/${runId}/results` : `/jobs/${jobId}/results`; return this.transport.requestJson("GET", path, { query }); } @@ -343,17 +378,38 @@ export class ZenRowsBatchClient { } } - /** Returns the scraped page's raw content as-is (HTML/text, not JSON). */ + /** + * Returns a task's stored content decoded as text (HTML, Markdown, plain text or JSON). + * Text decoding corrupts binary bodies: for a `pdf` or other binary result, use + * `getTaskContentBytes` instead. + * + * Throws a `ZenRowsBatchError`: 409 `task_not_ready` while the task runs, 422 with the + * task's error for a failed task, and 410 `result_expired` past the retention window. + */ async getTaskContent( jobId: string, taskId: string, options: { runId?: string } = {}, ): Promise { - const path = options.runId + return this.transport.requestText("GET", this.contentPath(jobId, taskId, options)); + } + + /** + * Returns a task's stored content as raw bytes, so binary results such as PDFs arrive + * intact. Same errors as `getTaskContent`. + */ + async getTaskContentBytes( + jobId: string, + taskId: string, + options: { runId?: string } = {}, + ): Promise { + return this.transport.requestBytes("GET", this.contentPath(jobId, taskId, options)); + } + + private contentPath(jobId: string, taskId: string, options: { runId?: string }): string { + return options.runId ? `/jobs/${jobId}/runs/${options.runId}/tasks/${taskId}/content` : `/jobs/${jobId}/tasks/${taskId}/content`; - const response = await this.transport.requestRaw("GET", path); - return response.text(); } getTaskHistory( @@ -407,12 +463,31 @@ export class ZenRowsBatchClient { return this.transport.requestJson("GET", `/jobs/${jobId}/webhook`); } - putJobWebhook(jobId: string, config: BatchWebhookConfig): Promise { + /** + * Replace a job's webhook. The API needs both `url` and `signature` on this call, so pass + * `signature` whenever you know the setting you want. + * + * When `signature` is left out, the SDK reads the job's current setting first (`false` when + * the job has no webhook yet) and sends that, so changing only the URL never turns signing + * off. That is two requests, not one atomic update: a change someone else makes in between + * is overwritten with the value read. + */ + async putJobWebhook(jobId: string, config: BatchWebhookConfig): Promise { + const signature = config.signature ?? (await this.currentWebhookSignature(jobId)); return this.transport.requestJson("PUT", `/jobs/${jobId}/webhook`, { - body: { signature: false, ...config }, + body: { url: config.url, signature }, }); } + private async currentWebhookSignature(jobId: string): Promise { + try { + return (await this.getJobWebhook(jobId)).signature === true; + } catch (error) { + if (error instanceof ZenRowsBatchError && error.status === 404) return false; + throw error; + } + } + async deleteJobWebhook(jobId: string): Promise { await this.transport.requestJson("DELETE", `/jobs/${jobId}/webhook`); } @@ -526,7 +601,7 @@ export class ZenRowsBatchClient { }); } - /** Start an export, wait for it, and save the zip to `targetPath`. Capped at 1 GiB per run. */ + /** Start an export, wait for it, and save the zip to `targetPath`. The API caps the size of one export. */ async downloadAllResults( jobId: string, runId: string, @@ -633,10 +708,11 @@ export class ZenRowsBatchClient { async _postTasks( jobId: string, tasks: BatchTask[], - options: { lastBatch?: boolean } = {}, + options: { lastBatch?: boolean; idempotencyKey?: string } = {}, ): Promise { return this.transport.requestJson("POST", `/jobs/${jobId}/tasks`, { body: { tasks, last_batch: options.lastBatch ?? false }, + idempotencyKey: options.idempotencyKey, }); } @@ -653,13 +729,26 @@ export class ZenRowsBatchClient { /** @internal */ async _waitForRunRaw(jobId: string, options: WaitForRunOptions): Promise { const target = options.targetStatuses ?? TERMINAL_RUN_STATUSES; - const fetchRun = async (): Promise => { + const loadRun = async (): Promise => { if (!options.runId) { const job = await this._getJobData(jobId); return job.latest_run; } return this._getRunData(jobId, options.runId); }; + const fetchRun = async (): Promise => { + const run = await loadRun(); + if ( + run !== undefined && + run.pause_state === "paused" && + !target.has(run.status) && + !options.failureStatuses?.has(run.status) && + !options.waitWhilePaused + ) { + throw new RunPausedError(run); + } + return run; + }; return pollUntil(fetchRun, { // A status the caller named as a failure throws, even when it is also a target. isDone: (run) => @@ -677,12 +766,29 @@ export class ZenRowsBatchClient { jobId: string, options: { timeout?: number; pollInterval?: number; maxPollInterval?: number } = {}, ): Promise { + // Ingestion is done only when the API says so. A run stopped mid-ingest is already + // terminal while its rows and totals are still being settled, so the run status alone + // is not enough. return pollUntil(() => this._getJobData(jobId), { - isDone: (job) => { - const run = job.latest_run; - if (!run || TERMINAL_RUN_STATUSES.has(run.status)) return true; - return run.ingest_status !== "pending"; - }, + isDone: (job) => !job.latest_run || job.latest_run.ingest_status !== "pending", + timeout: options.timeout ?? 300, + initialInterval: options.pollInterval ?? 2, + maxInterval: options.maxPollInterval ?? 15, + }); + } + + /** + * @internal Wait for one run's ingestion, by run id. A replayed submit or rerun returns + * the original run, which may no longer be the job's latest, so this never follows + * `latest_run`. + */ + async _waitForRunIngestRaw( + jobId: string, + runId: string, + options: { timeout?: number; pollInterval?: number; maxPollInterval?: number } = {}, + ): Promise { + return pollUntil(() => this._getRunData(jobId, runId), { + isDone: (run) => run.ingest_status !== "pending", timeout: options.timeout ?? 300, initialInterval: options.pollInterval ?? 2, maxInterval: options.maxPollInterval ?? 15, @@ -692,3 +798,4 @@ export class ZenRowsBatchClient { export { CurrentRun, ExportHandle, ExportRef, JobHandle, JobRef, RunHandle, RunRef }; export type { DownloadedResult }; +export * from "./webhook.js"; diff --git a/src/batch/download.ts b/src/batch/download.ts index 73ef72c..d52bdf9 100644 --- a/src/batch/download.ts +++ b/src/batch/download.ts @@ -17,13 +17,63 @@ export interface DownloadedResult { body: Buffer; } +/** + * A result row a bulk download left out because it has no body to fetch: an `expired` row + * (a successful result past the retention window) or a `failed` task. Reported through the + * `onSkip` option of `downloadToDir` / `downloadToMemory`. + */ +export interface SkippedResult { + taskId: string; + externalId?: string; + url: string; + status: "expired" | "failed"; +} + +/** Statuses whose rows have no stored body by design, so a bulk download skips them. */ +const SKIPPABLE_STATUSES = new Set(["expired", "failed"]); + +/** + * Split `results` into rows to fetch and rows to skip, enforcing `limit`. Only `expired` + * and `failed` rows without a `result_url` are skipped; any other row without one (a + * `successful` row missing its URL, or an unfinished task) stays in, so fetching it fails + * loudly instead of being dropped. + */ +async function partition( + results: AsyncIterable, + limit: number, + limitName: string, + onSkip: ((row: SkippedResult) => void) | undefined, +): Promise { + const fetchable: TaskResult[] = []; + for await (const task of results) { + if (!task.result_url && SKIPPABLE_STATUSES.has(task.status)) { + onSkip?.({ + taskId: task.task_id, + externalId: task.external_id, + url: task.url, + status: task.status as SkippedResult["status"], + }); + continue; + } + fetchable.push(task); + if (fetchable.length > limit) { + throw new Error(`result count exceeds ${limitName} (${limit})`); + } + } + return fetchable; +} + /** * Every result body lives at a presigned `result_url` — no auth header, no API content * endpoint in the loop, matching the Python SDK's approach exactly. */ async function fetchResultBody(task: TaskResult, maxBytes?: number): Promise { if (!task.result_url) { - throw new Error(`task ${task.task_id} has no result_url (not a successful task?)`); + const why = + task.status === "expired" + ? "its result is past the retention window (status expired)" + : `it has no stored result (status ${task.status})`; + throw new Error(`task ${task.task_id} has no result_url: ${why}`); } const response = await fetch(task.result_url); if (!response.ok) { @@ -75,9 +125,16 @@ export interface DownloadToDirOptions { concurrency?: number; maxFiles?: number; maxBytesPerFile?: number; + /** Called for each `expired` or `failed` row skipped because it has no body to fetch. */ + onSkip?: (row: SkippedResult) => void; } -/** Write every task body in `results` to `targetDir`, one file per task. */ +/** + * Write every task body in `results` to `targetDir`, one file per task, and return how many + * files it wrote. `expired` and `failed` rows have no body: they are skipped (and passed to + * `onSkip`) instead of failing the download. Any other row without a `result_url` still + * fails it. + */ export async function downloadToDir( results: AsyncIterable, targetDir: string, @@ -87,16 +144,11 @@ export async function downloadToDir( concurrency = 1, maxFiles = DEFAULT_MAX_FILES, maxBytesPerFile = DEFAULT_MAX_BYTES_PER_FILE, + onSkip, }: DownloadToDirOptions = {}, ): Promise { await mkdir(targetDir, { recursive: true }); - const tasks: TaskResult[] = []; - for await (const task of results) { - tasks.push(task); - if (tasks.length > maxFiles) { - throw new Error(`result count exceeds maxFiles (${maxFiles})`); - } - } + const tasks = await partition(results, maxFiles, "maxFiles", onSkip); await runPool(tasks, concurrency, async (task: TaskResult) => { const body = await fetchResultBody(task, maxBytesPerFile); @@ -114,9 +166,15 @@ export interface DownloadToMemoryOptions { maxCount?: number; maxTotalBytes?: number; maxBytesPerFile?: number; + /** Called for each `expired` or `failed` row skipped because it has no body to fetch. */ + onSkip?: (row: SkippedResult) => void; } -/** Load every task body in `results` into memory. */ +/** + * Load every task body in `results` into memory. `expired` and `failed` rows have no body: + * they are skipped (and passed to `onSkip`). Any other row without a `result_url` still + * fails the download. + */ export async function downloadToMemory( results: AsyncIterable, { @@ -124,15 +182,10 @@ export async function downloadToMemory( maxCount = DEFAULT_MAX_COUNT_IN_MEMORY, maxTotalBytes = DEFAULT_MAX_TOTAL_BYTES_IN_MEMORY, maxBytesPerFile = DEFAULT_MAX_BYTES_PER_FILE, + onSkip, }: DownloadToMemoryOptions = {}, ): Promise { - const tasks: TaskResult[] = []; - for await (const task of results) { - tasks.push(task); - if (tasks.length > maxCount) { - throw new Error(`result count exceeds maxCount (${maxCount})`); - } - } + const tasks = await partition(results, maxCount, "maxCount", onSkip); const downloaded: DownloadedResult[] = new Array(tasks.length); let totalBytes = 0; diff --git a/src/batch/errors.ts b/src/batch/errors.ts index 7a2e25b..d7417ee 100644 --- a/src/batch/errors.ts +++ b/src/batch/errors.ts @@ -1,17 +1,118 @@ import type { ProblemJson } from "./types.js"; +/** + * The `code` values the Batch API sends in an error body, by name. Compare + * `ZenRowsBatchError.code` (or a task's `error.code`) against these instead of a bare string. + * The API may add codes, so keep a default branch. + */ +export const BatchErrorCode = { + // Generic + /** 400. The request is invalid; a bad task is listed in `extras.invalid_tasks`. */ + InvalidArgument: "invalid_argument", + /** 404. */ + NotFound: "not_found", + /** 409. */ + Conflict: "conflict", + /** 401. The API key is missing or invalid. */ + Unauthenticated: "unauthenticated", + /** 401. The key is valid but the subscription is not active; fix it in Billing. */ + SubscriptionInactive: "subscription_inactive", + /** 402. The account is out of credits. */ + PaymentRequired: "payment_required", + /** 402. The API key reached one of its credit caps; `detail` names the cap and when it resets. */ + ApiKeyCapReached: "api_key_cap_reached", + /** 403. */ + PermissionDenied: "permission_denied", + /** + * 429. An account limit: active jobs (shared with Crawl), scheduled jobs, or the task cap of + * an open job. It clears only when a job finishes or is deleted, so it is not retried. + */ + QuotaExceeded: "quota_exceeded", + /** 500. */ + Internal: "internal", + /** 501. */ + NotImplemented: "not_implemented", + + // Job, run and task lifecycle (409 unless noted) + JobNotOpen: "job_not_open", + RunNotTerminal: "run_not_terminal", + RunNotStoppable: "run_not_stoppable", + NoPriorRun: "no_prior_run", + /** A rerun with a status filter matched no task. */ + NoMatchingTasks: "no_matching_tasks", + RunHasDescendants: "run_has_descendants", + LastBatchReceived: "last_batch_received", + /** The same `Idempotency-Key` was sent with a different body. */ + IdempotencyKeyConflict: "idempotency_key_conflict", + /** The task is still pending or processing, so it has no content yet. */ + TaskNotReady: "task_not_ready", + /** 410. The result is past the retention window. */ + ResultExpired: "result_expired", + /** A partial rerun (`status` filter) more than 10 days after the run started; a full rerun still works. */ + PartialRerunWindowExpired: "partial_rerun_window_expired", + + // HMAC signing keys (409) + CandidatePending: "candidate_pending", + NoCandidate: "no_candidate", + + // Input validation (400, usually per task in `invalid_tasks`) + UnsupportedScheme: "unsupported_scheme", + MalformedUrl: "malformed_url", + MissingHost: "missing_host", + UrlTooLong: "url_too_long", + MetadataTooLarge: "metadata_too_large", + MetadataKeyInvalid: "metadata_key_invalid", + InvalidExternalId: "invalid_external_id", + + // Task method and body (400) + UnsupportedMethod: "unsupported_method", + BodyNotAllowed: "body_not_allowed", + BodyTooLarge: "body_too_large", + BodyMalformed: "body_malformed", + MethodParamConflict: "method_param_conflict", + + // CSV file inputs + /** 404. */ + FileInputNotFound: "file_input_not_found", + /** 409. */ + FileInputNotUploaded: "file_input_not_uploaded", + /** 409. */ + FileInputUnsupportedForJobType: "file_input_unsupported_for_job_type", + + // Webhooks (400) + CallbackUrlMustBeHttps: "callback_url_must_be_https", + CallbackUrlUnresolvable: "callback_url_unresolvable", + WebhookSigningRequiresActiveKey: "webhook_signing_requires_active_key", + + // Task level: seen in a failed task's `error.code` + GatewayUnreachable: "gateway_unreachable", +} as const; + +/** One of the `BatchErrorCode` values. */ +export type BatchErrorCodeValue = (typeof BatchErrorCode)[keyof typeof BatchErrorCode]; + /** * A non-2xx response from the Batch API, decoded as RFC 7807 `application/problem+json` - * where possible. `code` is the stable Problem `code` member (e.g. `file_input_not_found`) — - * safe to branch on; defaults to `"internal"` when the body wasn't valid Problem JSON. + * where possible. `code` is the stable Problem `code` member (e.g. `file_input_not_found`), + * safe to branch on against `BatchErrorCode`; it is `"internal"` when the body wasn't valid + * Problem JSON. */ export class ZenRowsBatchError extends Error { readonly status: number; readonly problem: ProblemJson | undefined; readonly code: string; readonly extras: Record | undefined; + /** The Problem `detail`: a human-readable explanation, when the body had one. */ + readonly detail: string | undefined; + /** Seconds from the `Retry-After` header, when the response had one. */ + readonly retryAfter: number | undefined; - constructor(status: number, problem: ProblemJson | undefined, extras?: Record) { + constructor( + status: number, + problem: ProblemJson | undefined, + extras?: Record, + retryAfter?: number, + ) { const code = problem?.code ?? "internal"; const message = problem ? `${status} ${problem.title ?? "Error"}: ${problem.detail ?? code}` @@ -22,6 +123,8 @@ export class ZenRowsBatchError extends Error { this.problem = problem && { ...problem, code }; this.code = code; this.extras = extras; + this.detail = problem?.detail; + this.retryAfter = retryAfter; } } diff --git a/src/batch/estimate.ts b/src/batch/estimate.ts index 8848154..94ae1e0 100644 --- a/src/batch/estimate.ts +++ b/src/batch/estimate.ts @@ -21,8 +21,6 @@ const TIER_ORDER: Tier[] = [ "auto", ]; -const TRUTHY_STRINGS = new Set(["true", "1", "yes", "on"]); - export type ParamValue = string | boolean | number | Record; export type ParamMap = Record; @@ -46,19 +44,19 @@ export interface CostEstimate { exact: boolean; } +/** + * A flag is on exactly when the Batch API would read it as on: JSON `true`, or a string + * equal to "true" in any case. The API rejects other spellings ("1", "yes", numbers) with + * a 400 at submit, so they never bill; they count as off here. + */ function truthy(value: ParamValue | undefined): boolean { - if (typeof value === "boolean") return value; - if (typeof value === "number") return value !== 0; - if (typeof value === "string") return TRUTHY_STRINGS.has(value.trim().toLowerCase()); - return false; + if (value === true) return true; + return typeof value === "string" && value.toLowerCase() === "true"; } +/** The API accepts only the exact value "auto" for `mode`. */ function isAuto(params: ParamMap): boolean { - return ( - String(params.mode ?? "") - .trim() - .toLowerCase() === "auto" - ); + return params.mode === "auto"; } function costForParams(params: ParamMap): { tier: Tier; min: number; max: number } { @@ -91,8 +89,13 @@ function taskParams(task: TaskLike): ParamMap { /** * Estimate the credit cost of a job before submitting it. `tasks` accepts the same shapes - * `submitRegular` does: bare URL strings or task objects. Per-task `zenrows_params` override - * the job-level `zenrowsParams` on key collision (task wins), matching the worker's merge. + * `submitRegular` does: bare URL strings or task objects. Each task is priced on the + * job-level `zenrowsParams` overlaid with its own `zenrows_params` (the task wins on a key + * collision), the same merge the API applies. + * + * Scrape params set directly on a task object (a flat `js_render` next to `url`) are not + * counted: the API has no such task field and ignores it, so it doesn't change the price. + * Put per-task params under `zenrows_params`. */ export function estimateCost(tasks: Iterable, zenrowsParams?: ParamMap): CostEstimate { const jobParams = zenrowsParams ?? {}; diff --git a/src/batch/resources.ts b/src/batch/resources.ts index 96a50f5..9e39681 100644 --- a/src/batch/resources.ts +++ b/src/batch/resources.ts @@ -17,6 +17,7 @@ import type { StartExportResponse, SubmitJobResponse, TaskHistoryResponse, + TaskMethod, TaskResult, WebhookConfig, } from "./types.js"; @@ -26,6 +27,10 @@ interface BatchTaskLike { external_id?: string; metadata?: Record; zenrows_params?: Record; + /** HTTP method for the target fetch. Default `GET`. */ + method?: TaskMethod; + /** Request body for a `POST` task: a JSON value or a string, up to 16 KiB. */ + body?: unknown; } // ======================= specific runs ======================= @@ -96,6 +101,8 @@ export class RunRef { failureStatuses?: Set; timeout?: number; pollInterval?: number; + /** Keep polling while the run is paused instead of throwing `RunPausedError`. */ + waitWhilePaused?: boolean; } = {}, ): Promise { const run = await this.#client.waitForRun(this.jobId, { ...options, runId: this.runId }); @@ -270,6 +277,8 @@ export class CurrentRun { failureStatuses?: Set; timeout?: number; pollInterval?: number; + /** Keep polling while the run is paused instead of throwing `RunPausedError`. */ + waitWhilePaused?: boolean; } = {}, ): Promise { const run = await this.#client.waitForRun(this.jobId, options); @@ -415,24 +424,52 @@ export class JobRef { await this.#client._delete(this.jobId); } + /** + * Start a new run. Without `status` every task runs again; with it (for example + * `["failed", "pending"]`) only tasks in those states run and the rest are inherited. + * + * The API answers every rerun with 202 while it writes the new run's task rows, so + * `results()` read straight after a rerun can be partial. Pass `waitForIngest: true` to + * resolve only once ingestion is done (it polls up to 300 s), or call `waitForIngest()`. + */ async rerun( - options: { status?: string | string[]; idempotencyKey?: string } = {}, + options: { + status?: string | string[]; + idempotencyKey?: string; + waitForIngest?: boolean; + } = {}, ): Promise { - const resp = await this.#client._postRerun(this.jobId, options); - return new RunHandle(this.#client, this.jobId, resp.latest_run.run_id, resp.latest_run); + const { waitForIngest, ...rest } = options; + const resp = await this.#client._postRerun(this.jobId, rest); + let run = resp.latest_run; + if (waitForIngest && run.ingest_status === "pending") { + // By run id: a replayed rerun returns its original run, which may not be the latest. + run = await this.#client._waitForRunIngestRaw(this.jobId, run.run_id); + } + return new RunHandle(this.#client, this.jobId, run.run_id, run); } /** Shortcut for `rerun({ status: "failed" })` (or `"failed,pending"` with `includePending`). */ retryFailed( - options: { includePending?: boolean; idempotencyKey?: string } = {}, + options: { includePending?: boolean; idempotencyKey?: string; waitForIngest?: boolean } = {}, ): Promise { const status = options.includePending ? ["failed", "pending"] : "failed"; - return this.rerun({ status, idempotencyKey: options.idempotencyKey }); + return this.rerun({ + status, + idempotencyKey: options.idempotencyKey, + waitForIngest: options.waitForIngest, + }); } + /** + * Append tasks to an open job (up to 10,000 per call). With an `idempotencyKey`, a repeat of + * the same call (same key, same body) returns the original answer instead of adding the + * tasks again, so the SDK retries it on a transient failure. Without one, every call appends + * its tasks and a failed call is not retried. + */ addTasks( tasks: BatchTaskLike[], - options: { lastBatch?: boolean } = {}, + options: { lastBatch?: boolean; idempotencyKey?: string } = {}, ): Promise<{ accepted_tasks: number; job_status: Extensible; @@ -469,10 +506,25 @@ export class JobRef { await this.#client.deleteJobWebhook(this.jobId); } - /** Block until the current run's async-carrier ingestion has finished writing task rows. */ + /** + * Block until ingestion has finished writing the run's task rows (`ingest_status` is no + * longer `pending`), then return the loaded job. On a ref from a `submit*` call this waits + * for the run that submit returned; otherwise for the job's latest run. A run stopped + * mid-ingest still waits until the API has settled its rows. + */ async waitForIngest( options: { timeout?: number; pollInterval?: number; maxPollInterval?: number } = {}, ): Promise { + const runId = this.submitResponse?.latest_run?.run_id; + if (runId) { + await this.#client._waitForRunIngestRaw(this.jobId, runId, options); + return new JobHandle( + this.#client, + this.jobId, + await this.#client._getJobData(this.jobId), + this.submitResponse, + ); + } return new JobHandle( this.#client, this.jobId, diff --git a/src/batch/transport.ts b/src/batch/transport.ts index afd221a..26a5413 100644 --- a/src/batch/transport.ts +++ b/src/batch/transport.ts @@ -43,8 +43,27 @@ export type ErrorFactory = ( extras: Record | undefined, ) => Error; -const batchError: ErrorFactory = (response, problem, extras) => - new ZenRowsBatchError(response.status, problem, extras); +const batchError: ErrorFactory = (response, problem, extras) => { + const raw = response.headers.get("Retry-After"); + const seconds = raw ? Number(raw) : Number.NaN; + const retryAfter = Number.isFinite(seconds) && seconds >= 0 ? seconds : undefined; + return new ZenRowsBatchError(response.status, problem, extras, retryAfter); +}; + +/** + * The Batch API's 429 `quota_exceeded` is an account limit (active jobs, scheduled jobs, or + * an open job's task cap). It clears only when a job finishes or is deleted, so retrying it + * only adds delay. Reads a clone, so the body stays available for the error. + */ +async function isQuotaExceeded(response: Response): Promise { + if (response.status !== 429) return false; + try { + const body = (await response.clone().json()) as { code?: unknown } | null; + return body?.code === "quota_exceeded"; + } catch { + return false; + } +} export interface RequestOptions { query?: object; @@ -103,6 +122,20 @@ export class BatchTransport { } } + /** + * Send a request and return its body as bytes; throw the factory's error on non-2xx. The + * timeout covers reading the body. + */ + async requestBytes(method: string, path: string, options: RequestOptions = {}): Promise { + const { response, clear } = await this.send(method, path, options); + try { + await this.throwUnlessOk(response); + return Buffer.from(await response.arrayBuffer()); + } finally { + clear(); + } + } + /** * Send a request and return the raw `Response`, for bodies read as a stream. The timeout * stops once the headers of a 2xx arrive; an error body is still read within it. @@ -177,7 +210,12 @@ export class BatchTransport { throw error; } - if (idempotent && attempt < this.retries && retryStatuses.has(response.status)) { + if ( + idempotent && + attempt < this.retries && + retryStatuses.has(response.status) && + !(await isQuotaExceeded(response)) + ) { // An unread body holds its connection until garbage collection. response.body?.cancel().catch(() => {}); clear(); diff --git a/src/batch/types.ts b/src/batch/types.ts index 3a475b3..dde629a 100644 --- a/src/batch/types.ts +++ b/src/batch/types.ts @@ -10,10 +10,19 @@ export type JobType = "regular" | "scheduled"; export type JobStatus = "open" | "closed" | "deleted"; export type ScheduleState = "active" | "paused"; export type RunStatus = "pending" | "running" | "completed" | "failed" | "stopped" | "deleted"; -export type TaskStatus = "pending" | "processing" | "successful" | "failed"; +/** + * `expired`: a successful task whose result is past the retention window (14 days by + * default). The row has no `result_url`, and its content endpoint answers 410 + * `result_expired`. A `status: "successful"` filter still matches expired rows. + */ +export type TaskStatus = "pending" | "processing" | "successful" | "failed" | "expired"; export type PauseState = "active" | "paused"; export type IngestStatus = "pending" | "done"; export type ExportStatus = "pending" | "running" | "completed" | "failed"; +/** What started a run: `manual` (a submit or rerun through the API) or `scheduled` (a schedule fired). */ +export type RunTrigger = "manual" | "scheduled"; +/** HTTP method a task fetches its URL with. `GET` is the default. */ +export type TaskMethod = "GET" | "POST"; /** * Why a run auto-failed (`status == "failed"`). * - `insufficient_credits` — the account is out of credits. @@ -31,6 +40,7 @@ export interface Spend { cost: number; } +/** @deprecated Results listings never return per-task spend; see `TaskHistoryEvent.spend`. */ export interface TaskSpend { total: Spend; last_attempt: Spend; @@ -65,6 +75,8 @@ export interface Run { job_id: string; run_sequence: number; status: Extensible; + /** What started this run. */ + trigger?: Extensible; stats: RunStats; last_batch_received?: boolean; pause_state?: PauseState; @@ -142,12 +154,20 @@ export interface TaskResult { run_id: string; url: string; metadata?: Record; - method?: "GET" | "POST"; + method?: TaskMethod; status: Extensible; type?: string; result_url?: string; error?: ProblemJson; source_run_id?: string; + /** When the task row was created (RFC 3339). */ + created_at?: string; + /** When the task row last changed, for example when an attempt settled (RFC 3339). */ + updated_at?: string; + /** + * @deprecated Never set: results listings do not return per-task spend. Read per-attempt + * spend from `getTaskHistory()`, or the run total from `run.stats.spend`. + */ spend?: TaskSpend; } @@ -225,3 +245,30 @@ export interface TestWebhookResponse { error?: string; elapsed_ms: number; } + +/** Event types a Batch webhook delivers. `webhook.test` comes only from `testWebhook()`. */ +export type WebhookEventType = "run.completed" | "run.failed" | "webhook.test"; + +/** + * The JSON body of a Batch webhook delivery (`schema: "zenrows.webhook/v1"`). Verify the + * raw body with `verifyWebhookSignature()` before parsing it, and deduplicate deliveries on + * `event_id` (also sent as the `X-ZenRows-Event-Id` header): a delivery can repeat. + */ +export interface WebhookEvent { + schema: Extensible<"zenrows.webhook/v1">; + event_id: string; + event_type: Extensible; + /** When the event was queued (RFC 3339). */ + occurred_at: string; + job_id: string; + external_id?: string; + metadata?: Record; + run_id: string; + run_sequence: number; + /** The run's counters when the event was queued. */ + stats?: RunStats; + /** Set on `run.failed` only. */ + failure_reason?: Extensible; + /** Set on `run.failed` only. */ + failure_detail?: string; +} diff --git a/src/batch/waiters.ts b/src/batch/waiters.ts index 417f123..272d2e6 100644 --- a/src/batch/waiters.ts +++ b/src/batch/waiters.ts @@ -1,3 +1,5 @@ +import type { Run } from "./types.js"; + /** Raised when a waiter's `timeout` elapsed before the target state was reached. */ export class WaiterTimeoutError extends Error { constructor(message: string) { @@ -14,6 +16,21 @@ export class WaiterFailureError extends Error { } } +/** + * Raised by a run waiter when the run is paused (`pause_state: "paused"`) and not yet in a + * target status. A paused run makes no progress until it is resumed, so the waiter stops + * instead of polling until its timeout. `run` is the run as last read. + */ +export class RunPausedError extends Error { + readonly run: Run; + + constructor(run: Run) { + super(`waiter: run ${run.run_id} is paused; resume it with job.run.resume() and wait again`); + this.name = "RunPausedError"; + this.run = run; + } +} + export interface PollUntilOptions { isDone(value: T): boolean; isFailure?(value: T): boolean; diff --git a/src/batch/webhook.ts b/src/batch/webhook.ts new file mode 100644 index 0000000..78f7a76 --- /dev/null +++ b/src/batch/webhook.ts @@ -0,0 +1,103 @@ +import { createHmac, timingSafeEqual } from "node:crypto"; + +/** The request header that carries a signed delivery's signature. */ +export const WEBHOOK_SIGNATURE_HEADER = "X-Signature"; + +/** The request header that carries a delivery's event id. Deduplicate deliveries on it. */ +export const WEBHOOK_EVENT_ID_HEADER = "X-ZenRows-Event-Id"; + +/** How far, in seconds, a signature's timestamp may be from now by default. */ +export const DEFAULT_WEBHOOK_TOLERANCE_SECONDS = 300; + +/** The API issues every webhook signing key as 32 random bytes. */ +const WEBHOOK_KEY_BYTES = 32; + +/** + * Decode a base64 signing secret, or return undefined when it isn't one the API could have + * issued. Node's base64 decoder skips characters it doesn't recognise, so "!!!" or "====" + * decode to an empty buffer, and an HMAC keyed with that would accept a delivery anyone can + * sign. Require exactly 32 bytes that re-encode to the same text (ignoring `=` padding). + */ +function decodeSecret(secret: string): Buffer | undefined { + const key = Buffer.from(secret, "base64"); + if (key.length !== WEBHOOK_KEY_BYTES) return undefined; + const canonical = (text: string) => text.replace(/=+$/, ""); + if (canonical(key.toString("base64")) !== canonical(secret)) return undefined; + return key; +} + +export interface VerifyWebhookSignatureOptions { + /** + * Reject a delivery whose `t` is further than this from `now`, in seconds. Default 300. + * `0` turns the check off. Anything that isn't a finite number of 0 or more makes every + * delivery fail to verify. + */ + toleranceSeconds?: number; + /** + * The current time as a Unix timestamp in seconds. Defaults to the system clock. A value + * that isn't a finite number makes every delivery fail to verify. + */ + now?: number; +} + +function parseSignatureHeader(header: string): { t?: string; v1?: string; kid?: string } { + const parts: { t?: string; v1?: string; kid?: string } = {}; + for (const piece of header.split(",")) { + const index = piece.indexOf("="); + if (index <= 0) continue; + const key = piece.slice(0, index).trim(); + const value = piece.slice(index + 1).trim(); + if (key === "t" || key === "v1" || key === "kid") { + parts[key] = value; + } + } + return parts; +} + +/** + * Check a Batch webhook delivery's `X-Signature` header. + * + * The header reads `t=,v1=,kid=`. The signature covers + * `t`, a literal `.`, then the raw request body, keyed with the secret `rotateHmacKey()` + * returned. That secret is base64; this helper decodes it to the key bytes. + * + * Pass the body exactly as received, before any JSON parsing. `secrets` maps each key id to + * its base64 secret, so a receiver accepts both the old and the new key during a rotation; the + * header's `kid` picks the one to check. The comparison is constant-time. + * + * Returns `false` for a missing or malformed header, an unknown `kid`, a secret that isn't a + * 32-byte base64 key, a timestamp outside the tolerance, or a wrong signature. + * Deliveries can repeat: deduplicate on the `X-ZenRows-Event-Id` header. + */ +export function verifyWebhookSignature( + rawBody: string | Uint8Array, + signatureHeader: string | null | undefined, + secrets: Record, + options: VerifyWebhookSignatureOptions = {}, +): boolean { + const { t, v1, kid } = parseSignatureHeader(signatureHeader ?? ""); + // v1 is a hex SHA-256 digest: exactly 64 hex characters. Anything else is malformed + // (Buffer.from(hex) would otherwise drop a trailing odd nibble). + if (!t || !v1 || !kid || !/^\d+$/.test(t) || !/^[0-9a-fA-F]{64}$/.test(v1)) return false; + + const secret = Object.prototype.hasOwnProperty.call(secrets, kid) ? secrets[kid] : undefined; + if (!secret) return false; + + // A bad option fails closed: NaN compares false, so it would otherwise skip the age check. + const tolerance = options.toleranceSeconds ?? DEFAULT_WEBHOOK_TOLERANCE_SECONDS; + if (!Number.isFinite(tolerance) || tolerance < 0) return false; + if (options.now !== undefined && !Number.isFinite(options.now)) return false; + if (tolerance > 0) { + const now = options.now ?? Math.floor(Date.now() / 1000); + if (Math.abs(now - Number(t)) > tolerance) return false; + } + + const key = decodeSecret(secret); + if (!key) return false; + const body = typeof rawBody === "string" ? Buffer.from(rawBody, "utf8") : Buffer.from(rawBody); + const expected = createHmac("sha256", key) + .update(Buffer.concat([Buffer.from(`${t}.`, "utf8"), body])) + .digest(); + const given = Buffer.from(v1, "hex"); + return given.length === expected.length && timingSafeEqual(given, expected); +} diff --git a/tests/batch-client.test.ts b/tests/batch-client.test.ts index 9626478..71854ff 100644 --- a/tests/batch-client.test.ts +++ b/tests/batch-client.test.ts @@ -116,12 +116,90 @@ describe("ZenRowsBatchClient — webhooks", () => { const current = await client.getJobWebhook("job_123"); expect(current).toMatchObject({ url: "https://example.com/hook", signature: true }); - const replaced = await client.putJobWebhook("job_123", { url: "https://example.com/hook2" }); - expect(replaced).toMatchObject({ url: "https://example.com/hook2", signature: false }); + let sent: unknown; + server.use( + http.put("https://async.api.zenrows.com/v1/jobs/job_123/webhook", async ({ request }) => { + sent = await request.json(); + return HttpResponse.json({ url: "https://example.com/hook2", signature: true }); + }), + ); + const replaced = await client.putJobWebhook("job_123", { + url: "https://example.com/hook2", + signature: true, + }); + expect(sent).toEqual({ url: "https://example.com/hook2", signature: true }); + expect(replaced).toMatchObject({ url: "https://example.com/hook2", signature: true }); await expect(client.deleteJobWebhook("job_123")).resolves.toBeUndefined(); }); + test("putJobWebhook without signature keeps the job's current signing setting", async () => { + const sent: unknown[] = []; + let current: { url: string; signature: boolean } | undefined = { + url: "https://example.com/hook", + signature: true, + }; + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_wh/webhook", () => + current + ? HttpResponse.json(current) + : HttpResponse.json( + { type: "about:blank", title: "Not found", status: 404, code: "not_found" }, + { status: 404, headers: { "Content-Type": "application/problem+json" } }, + ), + ), + http.put("https://async.api.zenrows.com/v1/jobs/job_wh/webhook", async ({ request }) => { + const body = (await request.json()) as { url: string; signature: boolean }; + sent.push(body); + return HttpResponse.json(body); + }), + ); + + // Signed webhook, URL changed without saying `signature`: signing stays on. + await client.putJobWebhook("job_wh", { url: "https://example.com/new" }); + // No webhook yet: the API's submit-time default, unsigned. + current = undefined; + await client.putJobWebhook("job_wh", { url: "https://example.com/first" }); + // An explicit value is sent as given. + await client.putJobWebhook("job_wh", { url: "https://example.com/x", signature: false }); + + expect(sent).toEqual([ + { url: "https://example.com/new", signature: true }, + { url: "https://example.com/first", signature: false }, + { url: "https://example.com/x", signature: false }, + ]); + }); + + test("putJobWebhook without signature is a read then a write, not an atomic update", async () => { + // Documents a known limit: another caller turns signing on between this call's GET and + // its PUT, and the PUT writes back the value it read. Pass `signature` to avoid it. + const state = { url: "https://example.com/hook", signature: false }; + let gets = 0; + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_race/webhook", () => { + gets += 1; + const seen = { ...state }; + state.signature = true; // the concurrent update lands right after this read + return HttpResponse.json(seen); + }), + http.put("https://async.api.zenrows.com/v1/jobs/job_race/webhook", async ({ request }) => { + Object.assign(state, await request.json()); + return HttpResponse.json(state); + }), + ); + + await client.putJobWebhook("job_race", { url: "https://example.com/new" }); + expect(state).toEqual({ url: "https://example.com/new", signature: false }); + expect(gets).toBe(1); + + // With `signature` passed, there is no read, so nothing to race with. + for (const signature of [false, true]) { + await client.putJobWebhook("job_race", { url: "https://example.com/new", signature }); + expect(state.signature).toBe(signature); + } + expect(gets).toBe(1); + }); + test("dispatches a synthetic test event", async () => { const result = await client.testWebhook({ url: "https://example.com/hook" }); expect(result).toMatchObject({ delivered: true, status_code: 200 }); @@ -327,3 +405,139 @@ describe("ZenRowsBatchClient — current-run pause/resume", () => { expect(resumed.data.pause_state).toBe("active"); }); }); + +describe("ZenRowsBatchClient: content, add-tasks idempotency, results paging, timeout", () => { + test("getTaskContentBytes returns the bytes untouched, getTaskContent returns text", async () => { + const pdf = Buffer.from([0x25, 0x50, 0x44, 0x46, 0xff, 0x00, 0xe2, 0x9c]); + server.use( + http.get( + "https://async.api.zenrows.com/v1/jobs/job_pdf/tasks/t1/content", + () => new HttpResponse(pdf, { headers: { "Content-Type": "application/pdf" } }), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const bytes = await client.getTaskContentBytes("job_pdf", "t1"); + expect(Buffer.isBuffer(bytes)).toBe(true); + expect(bytes.equals(pdf)).toBe(true); + + const text = await client.getTaskContent("job_123", "task_1"); + expect(text).toBe("latest run content"); + }); + + test("addTasks sends Idempotency-Key and is retried on a transient failure", async () => { + const keys: (string | null)[] = []; + server.use( + http.post("https://async.api.zenrows.com/v1/jobs/job_open/tasks", ({ request }) => { + keys.push(request.headers.get("Idempotency-Key")); + if (keys.length === 1) return new HttpResponse(null, { status: 503 }); + return HttpResponse.json({ + accepted_tasks: 1, + job_status: "open", + latest_run: { run_id: "r", job_id: "job_open", run_sequence: 1, status: "running" }, + }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const resp = await client + .job("job_open") + .addTasks([{ url: "https://example.com" }], { idempotencyKey: "batch-7" }); + expect(resp.accepted_tasks).toBe(1); + expect(keys).toEqual(["batch-7", "batch-7"]); + }); + + test("addTasks without a key is not retried", async () => { + let calls = 0; + server.use( + http.post("https://async.api.zenrows.com/v1/jobs/job_open/tasks", () => { + calls += 1; + return new HttpResponse(null, { status: 503 }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect( + client.job("job_open").addTasks([{ url: "https://example.com" }]), + ).rejects.toBeInstanceOf(ZenRowsBatchError); + expect(calls).toBe(1); + }); + + test("getResults does not send limit, which the results endpoint ignores", async () => { + let query: URLSearchParams | undefined; + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_q/results", ({ request }) => { + query = new URL(request.url).searchParams; + return HttpResponse.json({ results: [], next_cursor: null }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await client.getResults("job_q", { limit: 50, status: "failed", cursor: "c1" }); + expect(query?.has("limit")).toBe(false); + expect(query?.get("status")).toBe("failed"); + expect(query?.get("cursor")).toBe("c1"); + }); + + test("requests time out after 30 seconds unless `timeout` says otherwise", () => { + // biome-ignore lint/suspicious/noExplicitAny: reading the private transport to prove it + const timeoutOf = (client: ZenRowsBatchClient) => (client as any).transport.timeoutMs; + expect(timeoutOf(new ZenRowsBatchClient("API_KEY"))).toBe(30_000); + expect(timeoutOf(new ZenRowsBatchClient("API_KEY", { timeout: 5 }))).toBe(5_000); + }); + + test("a request that takes longer than `timeout` seconds is aborted", async () => { + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_slow", async () => { + await new Promise((resolve) => setTimeout(resolve, 500)); + return HttpResponse.json({ job_id: "job_slow" }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY", { timeout: 0.05, retries: 0 }); + await expect(client.getJob("job_slow")).rejects.toThrow(/timed out/); + }); +}); + +describe("ZenRowsBatchClient: task inputs, run-scoped content, webhook read errors", () => { + test("submitRegular sends string and object tasks, including method and body", async () => { + let sent: { tasks?: unknown[] } | undefined; + server.use( + http.post("https://async.api.zenrows.com/v1/jobs", async ({ request }) => { + sent = (await request.json()) as { tasks?: unknown[] }; + return HttpResponse.json({ job_id: "job_in", status: "closed", accepted_tasks: 2 }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await client.submitRegular([ + "https://example.com/a", + { url: "https://example.com/search", method: "POST", body: { q: "shoes" } }, + ]); + expect(sent?.tasks).toEqual([ + { url: "https://example.com/a" }, + { url: "https://example.com/search", method: "POST", body: { q: "shoes" } }, + ]); + }); + + test("getTaskContentBytes reads an explicit run's content", async () => { + const client = new ZenRowsBatchClient("API_KEY"); + const bytes = await client.getTaskContentBytes("job_123", "task_1", { runId: "run_1" }); + expect(bytes.toString("utf8")).toBe("run_1 content"); + }); + + test("putJobWebhook without signature surfaces a read error other than 404", async () => { + let puts = 0; + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_denied/webhook", () => + HttpResponse.json( + { type: "about:blank", title: "Forbidden", status: 403, code: "permission_denied" }, + { status: 403, headers: { "Content-Type": "application/problem+json" } }, + ), + ), + http.put("https://async.api.zenrows.com/v1/jobs/job_denied/webhook", () => { + puts += 1; + return HttpResponse.json({}); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect( + client.putJobWebhook("job_denied", { url: "https://example.com/hook" }), + ).rejects.toMatchObject({ status: 403 }); + expect(puts).toBe(0); + }); +}); diff --git a/tests/batch-consistency.test.ts b/tests/batch-consistency.test.ts new file mode 100644 index 0000000..4c8287f --- /dev/null +++ b/tests/batch-consistency.test.ts @@ -0,0 +1,437 @@ +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import { http, HttpResponse } from "msw"; +import { describe, expect, expectTypeOf, test } from "vitest"; +import { + At, + BatchErrorCode, + type BatchErrorCodeValue, + type Run, + RunPausedError, + type TaskResult, + type TaskStatus, + WaiterTimeoutError, + type WebhookEvent, + ZenRowsBatchClient, + ZenRowsBatchError, +} from "../src"; +import { server } from "./_setup"; + +const API = "https://async.api.zenrows.com/v1"; + +function problem(status: number, code: string, headers: Record = {}) { + return HttpResponse.json( + { type: `https://docs.zenrows.com/api-error-codes#${code}`, title: "Error", status, code }, + { status, headers: { "Content-Type": "application/problem+json", ...headers } }, + ); +} + +describe("429 quota_exceeded", () => { + test("is an account limit, so it is not retried", async () => { + let calls = 0; + server.use( + http.get(`${API}/jobs/job_quota`, () => { + calls += 1; + return problem(429, "quota_exceeded"); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const error = await client.getJob("job_quota").catch((e) => e); + expect(error).toBeInstanceOf(ZenRowsBatchError); + expect(error.code).toBe(BatchErrorCode.QuotaExceeded); + expect(calls).toBe(1); + }); + + test("is not retried on a submit that carries an Idempotency-Key either", async () => { + let calls = 0; + server.use( + http.post(`${API}/jobs`, () => { + calls += 1; + return problem(429, "quota_exceeded"); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect( + client.submitRegular(["https://example.com"], undefined, { idempotencyKey: "k1" }), + ).rejects.toMatchObject({ status: 429, code: "quota_exceeded" }); + expect(calls).toBe(1); + }); + + test("another 429 is still retried", async () => { + let calls = 0; + server.use( + http.get(`${API}/jobs/job_busy`, () => { + calls += 1; + if (calls === 1) return problem(429, "too_many_requests", { "Retry-After": "0" }); + return HttpResponse.json({ job_id: "job_busy", type: "regular", status: "closed" }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect(client.getJob("job_busy")).resolves.toMatchObject({ jobId: "job_busy" }); + expect(calls).toBe(2); + }); +}); + +describe("ZenRowsBatchError fields", () => { + test("exposes detail and Retry-After seconds", async () => { + server.use( + http.get(`${API}/jobs/job_detail`, () => + HttpResponse.json( + { + type: "https://docs.zenrows.com/api-error-codes#api_key_cap_reached", + title: "Payment Required", + status: 402, + code: "api_key_cap_reached", + detail: "monthly cap reached", + }, + { + status: 402, + headers: { "Content-Type": "application/problem+json", "Retry-After": "120" }, + }, + ), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const error = (await client.getJob("job_detail").catch((e) => e)) as ZenRowsBatchError; + expect(error.code).toBe(BatchErrorCode.ApiKeyCapReached); + expect(error.detail).toBe("monthly cap reached"); + expect(error.retryAfter).toBe(120); + }); + + test("BatchErrorCode names the codes the API sends", () => { + expect(BatchErrorCode.SubscriptionInactive).toBe("subscription_inactive"); + expect(BatchErrorCode.PaymentRequired).toBe("payment_required"); + expect(BatchErrorCode.ResultExpired).toBe("result_expired"); + expect(BatchErrorCode.PartialRerunWindowExpired).toBe("partial_rerun_window_expired"); + expect(BatchErrorCode.IdempotencyKeyConflict).toBe("idempotency_key_conflict"); + // Every value is unique and snake_case. + const values = Object.values(BatchErrorCode); + expect(new Set(values).size).toBe(values.length); + for (const value of values) expect(value).toMatch(/^[a-z]+(_[a-z]+)*$/); + expectTypeOf<"quota_exceeded">().toMatchTypeOf(); + }); +}); + +describe("content of an expired result", () => { + test("getTaskContent throws 410 result_expired", async () => { + server.use( + http.get(`${API}/jobs/job_old/tasks/t1/content`, () => problem(410, "result_expired")), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect(client.getTaskContent("job_old", "t1")).rejects.toMatchObject({ + status: 410, + code: BatchErrorCode.ResultExpired, + }); + }); +}); + +describe("waiting for ingestion", () => { + function runBody(runId: string, status: string, ingest: string) { + return { + run_id: runId, + job_id: "job_rr", + run_sequence: 2, + status, + ingest_status: ingest, + stats: { total: 3, completed: 0, successful: 0, failed: 0 }, + }; + } + + test("rerun({ waitForIngest: true }) waits on the returned run, not the job's latest", async () => { + // A replayed rerun returns its original run (run_2) while run_3 is now the latest and + // still ingesting. The wait must follow run_2. + const runReads: string[] = []; + let jobReads = 0; + server.use( + http.post(`${API}/jobs/job_rr/rerun`, () => + HttpResponse.json( + { + job_id: "job_rr", + status: "closed", + latest_run: runBody("run_2", "pending", "pending"), + retried_tasks: 3, + inherited_tasks: 0, + }, + { status: 202 }, + ), + ), + http.get(`${API}/jobs/job_rr/runs/:runId`, ({ params }) => { + runReads.push(String(params.runId)); + return HttpResponse.json(runBody(String(params.runId), "completed", "done")); + }), + http.get(`${API}/jobs/job_rr`, () => { + jobReads += 1; + return HttpResponse.json({ + job_id: "job_rr", + type: "regular", + status: "closed", + latest_run: runBody("run_3", "running", "pending"), + }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const run = await client + .job("job_rr") + .retryFailed({ idempotencyKey: "replayed", waitForIngest: true }); + expect(run.runId).toBe("run_2"); + expect(run.data.ingest_status).toBe("done"); + expect(runReads).toEqual(["run_2"]); + expect(jobReads).toBe(0); + + // Without the option, no extra read. + await client.job("job_rr").rerun(); + expect(runReads).toEqual(["run_2"]); + }); + + test("a run stopped mid-ingest is not done until ingest_status leaves pending", async () => { + let reads = 0; + server.use( + http.post(`${API}/jobs/job_rr/rerun`, () => + HttpResponse.json( + { + job_id: "job_rr", + status: "closed", + latest_run: runBody("run_2", "pending", "pending"), + retried_tasks: 3, + inherited_tasks: 0, + }, + { status: 202 }, + ), + ), + http.get(`${API}/jobs/job_rr/runs/run_2`, () => { + reads += 1; + // Stopped at once, but the API settles the rows and totals a poll later. + return HttpResponse.json(runBody("run_2", "stopped", reads < 2 ? "pending" : "done")); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + // The rerun path uses the default 2 s poll; drive the same wait directly with a short one. + const resp = await client._postRerun("job_rr", {}); + const run = await client._waitForRunIngestRaw("job_rr", resp.latest_run.run_id, { + pollInterval: 0.01, + }); + expect(run.status).toBe("stopped"); + expect(run.ingest_status).toBe("done"); + expect(reads).toBe(2); + }); + + test("JobRef.waitForIngest on a stopped run with pending ingest keeps polling", async () => { + let reads = 0; + server.use( + http.get(`${API}/jobs/job_stop_ingest`, () => { + reads += 1; + return HttpResponse.json({ + job_id: "job_stop_ingest", + type: "regular", + status: "closed", + latest_run: { + ...runBody("run_1", "stopped", reads < 2 ? "pending" : "done"), + job_id: "job_stop_ingest", + }, + }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const handle = await client.job("job_stop_ingest").waitForIngest({ pollInterval: 0.01 }); + expect(handle.data.latest_run?.ingest_status).toBe("done"); + expect(reads).toBe(2); + }); + + test("a replayed submit waits on the run it returned, then loads the job", async () => { + const runReads: string[] = []; + server.use( + http.post(`${API}/jobs`, () => + HttpResponse.json( + { + job_id: "job_big", + status: "closed", + accepted_tasks: 10_000, + latest_run: { ...runBody("run_1", "pending", "pending"), job_id: "job_big" }, + }, + { status: 202 }, + ), + ), + http.get(`${API}/jobs/job_big/runs/:runId`, ({ params }) => { + runReads.push(String(params.runId)); + return HttpResponse.json({ + ...runBody(String(params.runId), "completed", "done"), + job_id: "job_big", + }); + }), + http.get(`${API}/jobs/job_big`, () => + HttpResponse.json({ + job_id: "job_big", + type: "regular", + status: "closed", + latest_run: { ...runBody("run_3", "running", "pending"), job_id: "job_big" }, + }), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const job = await client.submitRegular(["https://example.com"], undefined, { + idempotencyKey: "replayed", + waitForIngest: true, + }); + expect(runReads).toEqual(["run_1"]); + expect(job.jobId).toBe("job_big"); + }); + + test("submitScheduled answers 201 with no run, so waitForIngest never polls", async () => { + let jobReads = 0; + server.use( + http.post(`${API}/jobs`, () => + HttpResponse.json( + { job_id: "job_sched", status: "closed", accepted_tasks: 1 }, + { status: 201 }, + ), + ), + http.get(`${API}/jobs/job_sched`, () => { + jobReads += 1; + return HttpResponse.json({ job_id: "job_sched", type: "scheduled", status: "closed" }); + }), + ); + const client = new ZenRowsBatchClient("API_KEY"); + const job = await client.submitScheduled( + new At("2030-01-01T00:00:00", "UTC"), + ["https://example.com"], + undefined, + { waitForIngest: true }, + ); + expect(job.jobId).toBe("job_sched"); + expect(jobReads).toBe(0); + }); +}); + +describe("content reads honour the request timeout", () => { + test("getTaskContent and getTaskContentBytes abort a body that stalls after the headers", async () => { + // A real socket: headers and a first chunk arrive, then the body never ends. + const sockets = new Set(); + const stalled = createServer((_req, res) => { + res.writeHead(200, { "Content-Type": "text/html" }); + res.write(""); + }); + stalled.on("connection", (socket) => sockets.add(socket)); + await new Promise((resolve) => stalled.listen(0, "127.0.0.1", resolve)); + const { port } = stalled.address() as AddressInfo; + try { + const client = new ZenRowsBatchClient("API_KEY", { + baseURL: `http://127.0.0.1:${port}/v1`, + timeout: 0.2, + retries: 0, + }); + await expect(client.getTaskContent("job_stall", "t1")).rejects.toThrow(/timed out/); + await expect(client.getTaskContentBytes("job_stall", "t1")).rejects.toThrow(/timed out/); + } finally { + for (const socket of sockets) socket.destroy(); + await new Promise((resolve) => stalled.close(resolve)); + } + }, 5_000); +}); + +describe("types follow the wire shapes", () => { + test("task, run and webhook event fields", () => { + expectTypeOf<"expired">().toMatchTypeOf(); + expectTypeOf().toEqualTypeOf(); + expectTypeOf().toEqualTypeOf(); + expectTypeOf<"scheduled">().toMatchTypeOf>(); + const event: WebhookEvent = { + schema: "zenrows.webhook/v1", + event_id: "evt_1", + event_type: "run.failed", + occurred_at: "2026-01-01T00:00:00Z", + job_id: "j1", + run_id: "r1", + run_sequence: 1, + failure_reason: "insufficient_credits", + }; + expect(event.event_type).toBe("run.failed"); + }); +}); + +describe("waiting on a paused run", () => { + function pausedJob(jobId: string, counter: { reads: number }) { + return http.get(`${API}/jobs/${jobId}`, () => { + counter.reads += 1; + return HttpResponse.json({ + job_id: jobId, + type: "regular", + status: "closed", + latest_run: { + run_id: "run_p", + job_id: jobId, + run_sequence: 1, + status: "running", + pause_state: "paused", + stats: { total: 2, completed: 1, successful: 1, failed: 0 }, + }, + }); + }); + } + + test("stops at once with RunPausedError carrying the run", async () => { + const counter = { reads: 0 }; + server.use(pausedJob("job_paused", counter)); + const client = new ZenRowsBatchClient("API_KEY"); + const error = await client + .job("job_paused") + .run.wait({ timeout: 5 }) + .catch((e) => e); + expect(error).toBeInstanceOf(RunPausedError); + expect((error as RunPausedError).run.pause_state).toBe("paused"); + expect((error as RunPausedError).run.run_id).toBe("run_p"); + expect(counter.reads).toBe(1); + }); + + test("a specific run's wait stops on pause too", async () => { + server.use( + http.get(`${API}/jobs/job_paused2/runs/run_p`, () => + HttpResponse.json({ + run_id: "run_p", + job_id: "job_paused2", + run_sequence: 1, + status: "pending", + pause_state: "paused", + stats: { total: 1, completed: 0, successful: 0, failed: 0 }, + }), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect(client.run("job_paused2", "run_p").wait()).rejects.toBeInstanceOf(RunPausedError); + }); + + test("waitWhilePaused keeps polling until the timeout", async () => { + const counter = { reads: 0 }; + server.use(pausedJob("job_paused3", counter)); + const client = new ZenRowsBatchClient("API_KEY"); + await expect( + client.waitForRun("job_paused3", { + waitWhilePaused: true, + timeout: 0.05, + pollInterval: 0.01, + }), + ).rejects.toBeInstanceOf(WaiterTimeoutError); + expect(counter.reads).toBeGreaterThan(1); + }); + + test("a paused run already in a target status resolves", async () => { + server.use( + http.get(`${API}/jobs/job_paused4`, () => + HttpResponse.json({ + job_id: "job_paused4", + type: "regular", + status: "closed", + latest_run: { + run_id: "run_p", + job_id: "job_paused4", + run_sequence: 1, + status: "stopped", + pause_state: "paused", + stats: { total: 1, completed: 1, successful: 1, failed: 0 }, + }, + }), + ), + ); + const client = new ZenRowsBatchClient("API_KEY"); + await expect(client.waitForRun("job_paused4")).resolves.toMatchObject({ status: "stopped" }); + }); +}); diff --git a/tests/batch-download.test.ts b/tests/batch-download.test.ts index b942c74..5bd9a5f 100644 --- a/tests/batch-download.test.ts +++ b/tests/batch-download.test.ts @@ -1,9 +1,10 @@ import { mkdtemp, readFile, readdir, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { http, HttpResponse } from "msw"; import { afterEach, beforeEach, describe, expect, test } from "vitest"; import { ZenRowsBatchClient } from "../src/batch/client"; -import "./_setup"; +import { server } from "./_setup"; describe("downloadToDir / downloadToMemory / single-task download", () => { const client = new ZenRowsBatchClient("API_KEY"); @@ -70,4 +71,158 @@ describe("downloadToDir / downloadToMemory / single-task download", () => { }), ).rejects.toThrow(/no result_url/); }); + + describe("rows without a result_url", () => { + // An `expired` row (past the retention window) still matches `status=successful` but has + // no result_url; a `failed` row has none either when the filter includes it. + const rows = (extra: object[] = []) => + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_mixed/results", () => + HttpResponse.json({ + results: [ + { + task_id: "task_a", + run_id: "r", + url: "https://example.com/a", + status: "successful", + result_url: "https://storage.example.test/body-a", + }, + { + task_id: "task_old", + external_id: "ext_old", + run_id: "r", + url: "https://example.com/old", + status: "expired", + }, + { + task_id: "task_bad", + run_id: "r", + url: "https://example.com/bad", + status: "failed", + }, + ...extra, + ], + next_cursor: null, + }), + ), + ); + + test("downloadToDir skips expired and failed rows and reports them to onSkip", async () => { + rows(); + const skipped: unknown[] = []; + const count = await client.downloadToDir("job_mixed", undefined, dir, { + status: "all", + onSkip: (row) => skipped.push(row), + }); + expect(count).toBe(1); + expect(skipped).toEqual([ + { + taskId: "task_old", + externalId: "ext_old", + url: "https://example.com/old", + status: "expired", + }, + { + taskId: "task_bad", + externalId: undefined, + url: "https://example.com/bad", + status: "failed", + }, + ]); + expect(await readdir(dir)).toEqual(["task_a"]); + }); + + test("downloadToMemory skips them and reports them to onSkip", async () => { + rows(); + const skipped: unknown[] = []; + const results = await client.downloadToMemory("job_mixed", undefined, { + status: "all", + onSkip: (row) => skipped.push(row), + }); + expect(results.map((r) => r.taskId)).toEqual(["task_a"]); + expect(skipped).toEqual([ + { + taskId: "task_old", + externalId: "ext_old", + url: "https://example.com/old", + status: "expired", + }, + { + taskId: "task_bad", + externalId: undefined, + url: "https://example.com/bad", + status: "failed", + }, + ]); + }); + + test("onSkip is optional", async () => { + rows(); + const results = await client.downloadToMemory("job_mixed", undefined, { status: "all" }); + expect(results.map((r) => r.taskId)).toEqual(["task_a"]); + }); + + test("a successful row with no result_url still fails the download", async () => { + rows([ + { task_id: "task_odd", run_id: "r", url: "https://example.com/odd", status: "successful" }, + ]); + const skipped: unknown[] = []; + await expect( + client.downloadToDir("job_mixed", undefined, dir, { + status: "all", + onSkip: (row) => skipped.push(row), + }), + ).rejects.toThrow(/task_odd has no result_url/); + expect(skipped).toHaveLength(2); + await expect( + client.downloadToMemory("job_mixed", undefined, { status: "all" }), + ).rejects.toThrow(/task_odd has no result_url/); + }); + + test("an unfinished row with no result_url still fails the download", async () => { + rows([ + { task_id: "task_wip", run_id: "r", url: "https://example.com/wip", status: "processing" }, + ]); + await expect( + client.downloadToMemory("job_mixed", undefined, { status: "all" }), + ).rejects.toThrow(/task_wip has no result_url/); + }); + + test("a single-task download of an expired row says why it has no body", async () => { + await expect( + client.downloadTaskToMemory({ + task_id: "task_old", + run_id: "r", + url: "https://example.com/old", + status: "expired", + }), + ).rejects.toThrow(/retention window/); + }); + }); + + test("maxFiles counts only the rows it will fetch, and still caps them", async () => { + server.use( + http.get("https://async.api.zenrows.com/v1/jobs/job_cap/results", () => + HttpResponse.json({ + results: [ + { task_id: "x", run_id: "r", url: "https://example.com/x", status: "expired" }, + ...["a", "b"].map((id) => ({ + task_id: id, + run_id: "r", + url: `https://example.com/${id}`, + status: "successful", + result_url: `https://storage.example.test/body-${id}`, + })), + ], + next_cursor: null, + }), + ), + ); + await expect(client.downloadToDir("job_cap", undefined, dir, { maxFiles: 1 })).rejects.toThrow( + "result count exceeds maxFiles (1)", + ); + await expect( + client.downloadToMemory("job_cap", undefined, { maxCount: 2 }), + ).resolves.toHaveLength(2); + }); }); diff --git a/tests/batch-estimate.test.ts b/tests/batch-estimate.test.ts index 30d03ae..5c724de 100644 --- a/tests/batch-estimate.test.ts +++ b/tests/batch-estimate.test.ts @@ -33,10 +33,46 @@ describe("estimateCost", () => { expect(result.breakdown[0]?.tier).toBe("auto"); }); - test("truthy string spellings ('1', 'yes', 'ON') count as on", () => { - expect(estimateCost([{ zenrows_params: { js_render: "yes" } }]).min).toBe(5); - expect(estimateCost([{ zenrows_params: { premium_proxy: "1" } }]).min).toBe(10); - expect(estimateCost([{ zenrows_params: { js_render: "ON" } }]).min).toBe(5); + test('a flag is on for JSON true or the string "true" in any case', () => { + expect(estimateCost([{ zenrows_params: { js_render: "true" } }]).min).toBe(5); + expect(estimateCost([{ zenrows_params: { js_render: "TRUE" } }]).min).toBe(5); + expect(estimateCost([{ zenrows_params: { premium_proxy: "True" } }]).min).toBe(10); + }); + + test("spellings the API rejects at submit count as off", () => { + // The API accepts only true/false for these flags and answers 400 for anything else, + // so none of these would ever bill as JS rendering or premium proxy. + for (const value of ["1", "yes", "on", " true", 1, 2]) { + expect(estimateCost([{ zenrows_params: { js_render: value } }]).min).toBe(1); + expect(estimateCost([{ zenrows_params: { premium_proxy: value } }]).min).toBe(1); + } + }); + + test("mode counts as auto only for the exact value the API accepts", () => { + expect(estimateCost([{ zenrows_params: { mode: "auto" } }]).max).toBe(25); + expect(estimateCost([{ zenrows_params: { mode: "AUTO" } }]).max).toBe(1); + }); + + test("scrape params set flat on a task are not counted (the API ignores them)", () => { + expect(estimateCost([{ url: "https://a.com", js_render: true }]).min).toBe(1); + expect(estimateCost([{ url: "https://a.com", premium_proxy: "true" }], {}).min).toBe(1); + }); + + test("job params, then the task's zenrows_params", () => { + const result = estimateCost( + [ + "https://a.com", + { url: "https://b.com", zenrows_params: { premium_proxy: true } }, + { url: "https://c.com", zenrows_params: { js_render: "false" } }, + ], + { js_render: true }, + ); + expect(result.breakdown).toEqual([ + { tier: "base", count: 1, unitMin: 1, unitMax: 1 }, + { tier: "js_render", count: 1, unitMin: 5, unitMax: 5 }, + { tier: "js_render+premium_proxy", count: 1, unitMin: 25, unitMax: 25 }, + ]); + expect(result.min).toBe(31); }); test("falsy string spellings and zero do not count as on", () => { diff --git a/tests/batch-webhook.test.ts b/tests/batch-webhook.test.ts new file mode 100644 index 0000000..9bdf518 --- /dev/null +++ b/tests/batch-webhook.test.ts @@ -0,0 +1,147 @@ +import { createHmac, randomBytes } from "node:crypto"; +import { describe, expect, test } from "vitest"; +import { WEBHOOK_EVENT_ID_HEADER, WEBHOOK_SIGNATURE_HEADER, verifyWebhookSignature } from "../src"; + +// A delivery signed the way the Batch API signs: HMAC-SHA256 over `t + "." + raw body`, keyed +// with the RAW key bytes, while the secret handed to the customer is those bytes in base64. +// The header was produced by a copy of the API's own Go signing code (fmt "%d.%s" into an +// HMAC, then "t=%d,v1=%s,kid=%s"), so the test does not just mirror the helper. +const SECRET_B64 = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="; +const T = 1760000000; +const BODY = + '{"schema":"zenrows.webhook/v1","event_id":"evt_1","event_type":"run.completed","job_id":"j1","run_id":"r1","run_sequence":1,"stats":{"total":2,"completed":2,"successful":2,"failed":0}}'; +const SIG = "1cb06c874dbfb275a921703f7d5ca5b649ad6543952534feab4cb3b111673b17"; +const HEADER = `t=${T},v1=${SIG},kid=k1`; +const secrets = { k1: SECRET_B64 }; + +describe("verifyWebhookSignature", () => { + test("accepts a genuine delivery (string or bytes body)", () => { + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: T })).toBe(true); + expect(verifyWebhookSignature(Buffer.from(BODY), HEADER, secrets, { now: T + 10 })).toBe(true); + }); + + test("rejects a tampered body or signature", () => { + expect( + verifyWebhookSignature(BODY.replace('"failed":0', '"failed":1'), HEADER, secrets, { now: T }), + ).toBe(false); + const flipped = `t=${T},v1=${SIG.slice(0, -1)}0,kid=k1`; + expect(verifyWebhookSignature(BODY, flipped, secrets, { now: T })).toBe(false); + }); + + test("rejects the mistake of keying the HMAC with the base64 text itself", () => { + const wrong = createHmac("sha256", SECRET_B64).update(`${T}.${BODY}`).digest("hex"); + expect(verifyWebhookSignature(BODY, `t=${T},v1=${wrong},kid=k1`, secrets, { now: T })).toBe( + false, + ); + }); + + test("picks the secret by kid, so the old and new key both verify during a rotation", () => { + const raw2 = randomBytes(32); + const sig2 = createHmac("sha256", raw2).update(`${T}.${BODY}`).digest("hex"); + const both = { k1: SECRET_B64, k2: raw2.toString("base64") }; + expect(verifyWebhookSignature(BODY, HEADER, both, { now: T })).toBe(true); + expect(verifyWebhookSignature(BODY, `t=${T},v1=${sig2},kid=k2`, both, { now: T })).toBe(true); + expect(verifyWebhookSignature(BODY, `t=${T},v1=${sig2},kid=k3`, both, { now: T })).toBe(false); + }); + + test("rejects a stale timestamp unless the tolerance is turned off", () => { + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: T + 301 })).toBe(false); + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: T - 301 })).toBe(false); + expect( + verifyWebhookSignature(BODY, HEADER, secrets, { now: T + 301, toleranceSeconds: 600 }), + ).toBe(true); + expect( + verifyWebhookSignature(BODY, HEADER, secrets, { now: T + 86_400, toleranceSeconds: 0 }), + ).toBe(true); + }); + + test("returns false for a malformed header", () => { + for (const header of [ + "", + "garbage", + `t=${T},kid=k1`, + `v1=${SIG},kid=k1`, + `t=abc,v1=${SIG},kid=k1`, + // A signature with one extra hex digit, or one missing, is not a SHA-256 digest. + `t=${T},v1=${SIG}f,kid=k1`, + `t=${T},v1=${SIG.slice(0, -1)},kid=k1`, + `t=${T},v1=${SIG.slice(0, -2)},kid=k1`, + ]) { + expect(verifyWebhookSignature(BODY, header, secrets, { now: T })).toBe(false); + } + }); + + test("rejects a delivery forged with an empty key when the secret decodes to nothing", () => { + // Node's base64 decoder drops characters it doesn't know, so these decode to zero bytes. + // An HMAC keyed with zero bytes is one anyone can compute. + const forgedBody = '{"forged":true}'; + const forged = createHmac("sha256", Buffer.alloc(0)).update(`${T}.${forgedBody}`).digest("hex"); + for (const bad of ["!!!", "====", "", " "]) { + expect( + verifyWebhookSignature(forgedBody, `t=${T},v1=${forged},kid=k1`, { k1: bad }, { now: T }), + ).toBe(false); + } + }); + + test("rejects a secret that is not exactly 32 bytes of canonical base64", () => { + const short = Buffer.alloc(16, 7); + const shortSig = createHmac("sha256", short).update(`${T}.${BODY}`).digest("hex"); + expect( + verifyWebhookSignature( + BODY, + `t=${T},v1=${shortSig},kid=k1`, + { k1: short.toString("base64") }, + { now: T }, + ), + ).toBe(false); + // The right 32 bytes, but with junk Node would silently skip: not the text the API issued. + expect(verifyWebhookSignature(BODY, HEADER, { k1: `${SECRET_B64}!!` }, { now: T })).toBe(false); + expect(verifyWebhookSignature(BODY, HEADER, { k1: `!${SECRET_B64}` }, { now: T })).toBe(false); + }); + + test("accepts the issued secret with or without its = padding", () => { + expect( + verifyWebhookSignature(BODY, HEADER, { k1: SECRET_B64.replace(/=+$/, "") }, { now: T }), + ).toBe(true); + }); + + test("invalid timestamp options fail closed instead of skipping the age check", () => { + const dayOld = T + 86_400; + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: dayOld })).toBe(false); + for (const toleranceSeconds of [ + -1, + Number.NaN, + Number.POSITIVE_INFINITY, + Number.NEGATIVE_INFINITY, + ]) { + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: dayOld, toleranceSeconds })).toBe( + false, + ); + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now: T, toleranceSeconds })).toBe( + false, + ); + } + for (const now of [Number.NaN, Number.POSITIVE_INFINITY, Number.NEGATIVE_INFINITY]) { + expect(verifyWebhookSignature(BODY, HEADER, secrets, { now })).toBe(false); + } + }); + + test("uses the system clock when `now` is not given", () => { + const t = Math.floor(Date.now() / 1000); + const raw = Buffer.from(SECRET_B64, "base64"); + const sig = createHmac("sha256", raw).update(`${t}.${BODY}`).digest("hex"); + expect(verifyWebhookSignature(BODY, `t=${t},v1=${sig},kid=k1`, secrets)).toBe(true); + // The fixed fixture is from 2025, far outside the default 5 minutes of the real clock. + expect(verifyWebhookSignature(BODY, HEADER, secrets)).toBe(false); + }); + + test("a missing header is rejected, not thrown", () => { + expect(verifyWebhookSignature(BODY, undefined, secrets, { now: T })).toBe(false); + expect(verifyWebhookSignature(BODY, null, secrets, { now: T })).toBe(false); + }); + + test("names the headers a delivery carries", () => { + expect(WEBHOOK_SIGNATURE_HEADER).toBe("X-Signature"); + expect(WEBHOOK_EVENT_ID_HEADER).toBe("X-ZenRows-Event-Id"); + }); +}); diff --git a/tests/readme-batch.test.ts b/tests/readme-batch.test.ts new file mode 100644 index 0000000..24540b1 --- /dev/null +++ b/tests/readme-batch.test.ts @@ -0,0 +1,70 @@ +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import ts from "typescript"; +import { describe, expect, test } from "vitest"; + +// The README's Batch examples are what people copy. Typecheck every ```ts block in the +// Batch section against the source, with `zenrows` resolving to src/index.ts, so an example +// that calls a method the client doesn't have fails CI. + +const ROOT = resolve(__dirname, ".."); + +function batchSnippets(): string[] { + const readme = readFileSync(join(ROOT, "README.md"), "utf8"); + const start = readme.indexOf("\n### Batch\n"); + const end = readme.indexOf("\n### Crawl\n", start); + if (start < 0 || end < 0) throw new Error("README has no Batch section before Crawl"); + const section = readme.slice(start, end); + return [...section.matchAll(/```ts\n([\s\S]*?)```/g)].map((match) => match[1] ?? ""); +} + +describe("README Batch section", () => { + test("every code block is TypeScript, so all of them get typechecked", () => { + const readme = readFileSync(join(ROOT, "README.md"), "utf8"); + const section = readme.slice(readme.indexOf("\n### Batch\n"), readme.indexOf("\n### Crawl\n")); + const fences = [...section.matchAll(/^```(\w*)$/gm)].map((match) => match[1]); + const openings = fences.filter((_, index) => index % 2 === 0); + expect(openings.length).toBeGreaterThan(0); + expect(openings.every((lang) => lang === "ts")).toBe(true); + }); + + test("its examples compile against the client", () => { + const snippets = batchSnippets(); + const dir = mkdtempSync(join(tmpdir(), "zenrows-readme-")); + try { + const files = snippets.map((code, index) => { + const file = join(dir, `snippet-${index}.ts`); + writeFileSync(file, code); + return file; + }); + const program = ts.createProgram(files, { + strict: true, + noEmit: true, + skipLibCheck: true, + target: ts.ScriptTarget.ES2022, + module: ts.ModuleKind.ESNext, + moduleResolution: ts.ModuleResolutionKind.Bundler, + moduleDetection: ts.ModuleDetectionKind.Force, + resolveJsonModule: true, + esModuleInterop: true, + lib: ["lib.es2022.d.ts"], + types: ["node"], + typeRoots: [join(ROOT, "node_modules", "@types")], + baseUrl: ROOT, + paths: { zenrows: ["src/index.ts"] }, + }); + const diagnostics = ts.getPreEmitDiagnostics(program).map((diagnostic) => { + const where = diagnostic.file + ? `${diagnostic.file.fileName.replace(dir, "README")}:${ + diagnostic.file.getLineAndCharacterOfPosition(diagnostic.start ?? 0).line + 1 + }` + : ""; + return `${where} ${ts.flattenDiagnosticMessageText(diagnostic.messageText, "\n")}`; + }); + expect(diagnostics).toEqual([]); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }, 60_000); +});