From 69a6138ba6da98aa7da2f8a186b5261b8d7e2e3f Mon Sep 17 00:00:00 2001 From: George Pickett Date: Mon, 14 Sep 2026 17:48:06 -0700 Subject: [PATCH] Fix webhook reliability and prepare a validated npm release --- .github/dependabot.yml | 11 + .github/workflows/ci.yaml | 15 +- .github/workflows/publish.yml | 16 +- .prettierignore | 5 + .prettierrc.js | 2 +- CONTRIBUTING.md | 49 ++++ README.md | 34 +++ credentials/ParallelApi.credentials.ts | 3 +- nodes/Parallel/Parallel.node.ts | 132 ++++++---- .../actions/asyncWebEnrichment.operation.ts | 17 +- .../actions/createMonitor.operation.ts | 50 +++- .../actions/deleteMonitor.operation.ts | 6 +- .../Parallel/actions/getMonitor.operation.ts | 6 +- .../actions/getMonitorEventGroup.operation.ts | 6 +- .../Parallel/actions/getTaskRun.operation.ts | 6 +- .../actions/getTaskRunResult.operation.ts | 6 +- .../actions/listMonitorEvents.operation.ts | 14 +- .../actions/listMonitors.operation.ts | 12 +- .../actions/triggerMonitor.operation.ts | 6 +- .../actions/updateMonitor.operation.ts | 14 +- nodes/Parallel/actions/webChat.operation.ts | 27 ++- .../actions/webEnrichment.operation.ts | 92 +++---- nodes/Parallel/actions/webSearch.operation.ts | 9 +- nodes/Parallel/contracts/requests.ts | 4 + nodes/Parallel/transport/ParallelApi.ts | 7 +- nodes/Parallel/transport/waitForTaskResult.ts | 49 ++++ nodes/Parallel/utils/index.ts | 6 +- nodes/Parallel/webhooks/response.ts | 32 +++ nodes/Parallel/webhooks/verify.ts | 11 +- .../ParallelMonitorTrigger.node.ts | 76 +++--- nodes/ParallelTrigger/ParallelTrigger.node.ts | 56 ++--- package.json | 10 +- scripts/check-package.mjs | 38 ++- scripts/release.mjs | 28 +++ test/reliability.test.js | 225 ++++++++++++++++++ test/webhooks.test.js | 2 + 36 files changed, 803 insertions(+), 279 deletions(-) create mode 100644 .github/dependabot.yml create mode 100644 .prettierignore create mode 100644 nodes/Parallel/transport/waitForTaskResult.ts create mode 100644 nodes/Parallel/webhooks/response.ts create mode 100644 scripts/release.mjs create mode 100644 test/reliability.test.js diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..e4c9f9a --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,11 @@ +version: 2 +updates: + - package-ecosystem: npm + directory: '/' + schedule: + interval: weekly + groups: + non-major: + update-types: + - minor + - patch diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index f0c9d19..0532cd8 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -12,6 +12,7 @@ jobs: validate: runs-on: ubuntu-latest strategy: + fail-fast: false matrix: node-version: ['22.22', '24'] steps: @@ -22,8 +23,12 @@ jobs: cache: npm - run: npm install --global npm@11.6.0 - run: npm ci - - run: npm run lint - - run: npm run typecheck - - run: npm test - - run: npm audit --omit=dev - - run: npm run check-package + - run: npm run validate + - name: Record full dependency audit + if: always() + run: npm audit --json > dependency-audit.json || test -s dependency-audit.json + - uses: actions/upload-artifact@v4 + if: always() + with: + name: dependency-audit-${{ matrix.node-version }} + path: dependency-audit.json diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index a2837b4..92df99d 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -3,11 +3,15 @@ name: Publish on: push: tags: - - '*.*.*' + - 'v*.*.*' permissions: contents: read +concurrency: + group: npm-publish + cancel-in-progress: false + jobs: publish: runs-on: ubuntu-latest @@ -16,6 +20,8 @@ jobs: id-token: write steps: - uses: actions/checkout@v5 + with: + fetch-depth: 0 - uses: actions/setup-node@v5 with: node-version: '24' @@ -23,9 +29,5 @@ jobs: registry-url: https://registry.npmjs.org - run: npm install --global npm@11.6.0 - run: npm ci - - run: npm run lint - - run: npm run typecheck - - run: npm test - - run: npm run check-package - - name: Publish with provenance - run: npm publish --provenance --access public + - name: Validate and publish with provenance + run: npm run release diff --git a/.prettierignore b/.prettierignore new file mode 100644 index 0000000..0ae60ca --- /dev/null +++ b/.prettierignore @@ -0,0 +1,5 @@ +node_modules/ +dist/ +.agent/ +package-lock.json +*.tsbuildinfo diff --git a/.prettierrc.js b/.prettierrc.js index f6ef3aa..f385577 100644 --- a/.prettierrc.js +++ b/.prettierrc.js @@ -9,4 +9,4 @@ module.exports = { quoteProps: 'as-needed', endOfLine: 'lf', printWidth: 100, -}; \ No newline at end of file +}; diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index b01acf3..5c63bf6 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -28,3 +28,52 @@ Tests use Node's built-in test runner against clean compiled output. Add contrac ## Pull requests Explain the user impact, compatibility boundary, and validation performed. Distinguish fixture/schema validation from authenticated live API validation. Do not publish a package from a pull-request branch. + +## Release checks + +Run `npm run validate` for the same checks as CI. Run `npm run release:rehearse` +to execute validation and publish lifecycle hooks without writing to npm. The +local package checker deliberately performs real temporary pack/install operations +even during this rehearsal. It also loads every declared node and credential. + +After an independent PR review, both CI jobs and the required Eng approval, merge +through the normal GitHub process. Finish the sequential dependency queue and +rerun the audit on final main before choosing an unpublished package version. +Update package.json and package-lock.json together if a version change is needed. +The reviewed release tag must be `v` followed by that version, on the tested main +commit. Never move a published tag or reuse a published version. + +Publishing runs only through `.github/workflows/publish.yml`, with npm trusted +publishing configured for this repository and workflow. An npm package owner must +verify that trust and accept any remaining development-tool security exposure. +A local npm login is not required by trusted publishing, but GitHub write access +alone does not establish npm publishing authority. Do not bypass lifecycle checks +or repository approval rules. Verify npm provenance, tarball integrity and a fresh +registry install after publishing. Rehearsal does not prove OIDC permission. + +The current dependency queue is #23, #20, #21, then #22. Refresh and test each PR +after the preceding merge. #22 subsumes the vulnerable brace-expansion path; +verify the final lockfile rather than assuming an overlapping PR is unnecessary. +Use `npm audit --json` to inspect the full tooling tree as well as the production +audit in validate. CI retains that report. Require a full-tree high/critical audit +gate once the existing queue clears; do not hide advisories or force unsupported +major overrides to produce a green audit. + +### Unresolved development dependency findings + +As of September 14, 2026, the production audit is clean but development tooling +still brings vulnerable `uuid@10.0.0`, `qs@6.15.2` and `stream-json@1.9.1`. +Track GHSA-w5hq-g745-h8pq, GHSA-x5fp-wj9c-mxmx, GHSA-4mjr-xmp4-gh2g and +GHSA-528h-pc64-c93x in the full audit report. The latest inspected n8n backend +packages still pin the affected qs and stream-json versions. This repository's +package validator prohibits overrides, including scoped overrides. A tested qs +6.16.0 override therefore cannot be shipped under the current package contract. + +The stream-json consumer imports its parser and `stream-json/Assembler` through +CommonJS; forcing version 3 would change the module/export contract. The uuid +consumers include LangChain v1/v4 usage in both CommonJS and ESM. These packages +are development dependencies, not libraries shipped in this node's tarball; +the separately installed n8n host has its own dependency tree. This distinction +is not risk acceptance. The release owner must disposition these findings or +wait for compatible upstream repairs before publishing. Reevaluate when the +n8n tooling updates its pinned dependencies; retain audit evidence in the PR. diff --git a/README.md b/README.md index 3bad28f..916a0df 100644 --- a/README.md +++ b/README.md @@ -81,3 +81,37 @@ Releases are published only by the tag-triggered GitHub Actions workflow. Config - [Monitor API](https://docs.parallel.ai/monitor-api/monitor-quickstart) - [Chat API](https://docs.parallel.ai/chat-api/chat-quickstart) - [n8n community-node documentation](https://docs.n8n.io/integrations/community-nodes/) + +## Upgrading from 0.2.0 + +Back up your n8n database and exported workflows before updating. Test the upgrade +in a disposable instance first, using Node 22.22 or newer. Existing node, +credential and operation identifiers remain unchanged. + +- Search defaults to Basic, equivalent to the previous Base setting. Explicit + Base and Pro values still map to Basic and Advanced. +- Webhook signature validation defaults to enabled. Workflows that omitted the + old default now require the Parallel webhook secret in their credentials. + Configure it before reactivating callbacks. An explicitly saved disabled setting + remains disabled. Both Standard and legacy signing formats are supported. +- Monitor callbacks retain `event_group_id` at the top level. Fetch failures still + emit `event_group_error` by default. Enable **Retry on Fetch Failure** to return + a retryable HTTP failure instead. Accepted events may be delivered repeatedly; + use the additive `webhook_id` to identify redelivery. With signature validation + disabled, this header is untrusted caller input. +- The GA Monitor events endpoint returns an `events` array and `next_cursor`. + Update expressions that consumed the old alpha event-group response accordingly. + Saved **Lookback Period** options are rejected explicitly because the current + API cannot guarantee that historical window. Remove that option and use cursor + pagination, handling unavailable older history in your workflow. +- Synchronous Tasks have a configurable total result-wait budget of 1–120 minutes, + defaulting to 75 minutes to accommodate the old 15 four-minute polls plus + backoff. This is now a hard local bound; host execution limits can be shorter. + A timeout does not cancel the Task. Use the reported run ID with **Get Task Run + Result**, rather than submitting another Task. Prefer Async for long processors. +- Authenticated events excluded by a trigger's filter receive HTTP 200 without + starting the workflow. Malformed matching events receive 400; invalid signatures + receive 401. Task result fetch failures and opted-in Monitor fetch failures + receive 503. Enrichment requests have a five-second local timeout. + +See [CONTRIBUTING.md](CONTRIBUTING.md) for validation and the reviewed release process. diff --git a/credentials/ParallelApi.credentials.ts b/credentials/ParallelApi.credentials.ts index ba4f8b8..e5f9ef2 100644 --- a/credentials/ParallelApi.credentials.ts +++ b/credentials/ParallelApi.credentials.ts @@ -35,7 +35,8 @@ export class ParallelApi implements ICredentialType { }, default: '', required: false, - description: 'Optional webhook secret for validating webhook signatures. Find this in Settings → Webhooks at https://platform.parallel.ai/settings', + description: + 'Optional webhook secret for validating webhook signatures. Find this in Settings → Webhooks at https://platform.parallel.ai/settings', hint: 'Webhook validation ensures secure communication. Get your secret from platform.parallel.ai/settings', }, ]; diff --git a/nodes/Parallel/Parallel.node.ts b/nodes/Parallel/Parallel.node.ts index 92f3cf3..5db8ec0 100644 --- a/nodes/Parallel/Parallel.node.ts +++ b/nodes/Parallel/Parallel.node.ts @@ -37,7 +37,8 @@ export class Parallel implements INodeType { icon: { light: 'file:parallel.svg', dark: 'file:parallel.dark.svg' }, group: ['transform'], version: 1, - subtitle: '={{$parameter["resource"] === "monitor" ? "Monitor / " + $parameter["monitorOperation"] : $parameter["operation"]}}', + subtitle: + '={{$parameter["resource"] === "monitor" ? "Monitor / " + $parameter["monitorOperation"] : $parameter["operation"]}}', description: 'Highest accuracy web search tools for AI agents', defaults: { name: 'Parallel', @@ -153,8 +154,10 @@ export class Parallel implements INodeType { inputType: ['json'], }, }, - default: '{\n "company_name": "Apple Inc.",\n "company_domain": "apple.com",\n "company_ticker": "AAPL"\n}', - description: 'System will expect inputs of this JSON structure. Provide actual data values here.', + default: + '{\n "company_name": "Apple Inc.",\n "company_domain": "apple.com",\n "company_ticker": "AAPL"\n}', + description: + 'System will expect inputs of this JSON structure. Provide actual data values here.', }, { displayName: 'Output Schema Type', @@ -206,7 +209,8 @@ export class Parallel implements INodeType { { name: 'Auto', value: 'auto', - description: 'Only supported in Pro and above. Optimized JSON output with nested citations.', + description: + 'Only supported in Pro and above. Optimized JSON output with nested citations.', }, ], default: 'text', @@ -224,7 +228,8 @@ export class Parallel implements INodeType { }, }, default: '', - placeholder: 'Optional: Describe the desired output (e.g., "Format as $X.X trillion (year)")', + placeholder: + 'Optional: Describe the desired output (e.g., "Format as $X.X trillion (year)")', description: 'Optional description of how you want the text output formatted', }, { @@ -241,7 +246,8 @@ export class Parallel implements INodeType { }, }, default: '', - placeholder: 'Required: Describe what text output you want from the JSON input (e.g., "Generate a company summary")', + placeholder: + 'Required: Describe what text output you want from the JSON input (e.g., "Generate a company summary")', description: 'Required description of what text output you want when providing JSON input', }, { @@ -261,7 +267,8 @@ export class Parallel implements INodeType { }, default: '{\n "type": "object",\n "properties": {\n "company_name": {\n "type": "string",\n "description": "Official company name from recent filings or website."\n },\n "ceo_name": {\n "type": "string",\n "description": "Current CEO full name from company website or recent news."\n },\n "employee_count": {\n "type": "string",\n "description": "Current number of employees as approximate number or range (e.g., \'500-1000\', \'2500\')."\n },\n "annual_revenue_2024": {\n "type": "string",\n "description": "2024 annual revenue in millions USD format (e.g., \'$500M\', \'$2.5B\')."\n },\n "headquarters_city": {\n "type": "string",\n "description": "Primary headquarters city and country (e.g., \'San Francisco, USA\')."\n },\n "founded_year": {\n "type": "string",\n "description": "Year company was founded in YYYY format."\n }\n },\n "required": ["company_name", "ceo_name", "employee_count", "annual_revenue_2024", "headquarters_city", "founded_year"],\n "additionalProperties": false\n}', - description: 'JSON schema defining the structure of the expected output (description fields serve as field-level prompts)', + description: + 'JSON schema defining the structure of the expected output (description fields serve as field-level prompts)', }, { displayName: 'JSON Schema', @@ -280,7 +287,8 @@ export class Parallel implements INodeType { }, default: '{\n "type": "object",\n "properties": {\n "company_name": {\n "type": "string",\n "description": "Official company name from recent filings or website."\n },\n "ceo_name": {\n "type": "string",\n "description": "Current CEO full name from company website or recent news."\n },\n "employee_count": {\n "type": "string",\n "description": "Current number of employees as approximate number or range (e.g., \'500-1000\', \'2500\')."\n },\n "annual_revenue_2024": {\n "type": "string",\n "description": "2024 annual revenue in millions USD format (e.g., \'$500M\', \'$2.5B\')."\n },\n "headquarters_city": {\n "type": "string",\n "description": "Primary headquarters city and country (e.g., \'San Francisco, USA\')."\n },\n "founded_year": {\n "type": "string",\n "description": "Year company was founded in YYYY format."\n }\n },\n "required": ["company_name", "ceo_name", "employee_count", "annual_revenue_2024", "headquarters_city", "founded_year"],\n "additionalProperties": false\n}', - description: 'JSON schema defining the structure of the expected output (required when JSON type is selected)', + description: + 'JSON schema defining the structure of the expected output (required when JSON type is selected)', }, { displayName: 'Processor', @@ -300,7 +308,8 @@ export class Parallel implements INodeType { displayName: 'Processor', name: 'asyncProcessor', type: 'options', - description: 'Processor used for the async task. Higher-end processors for longer-running tasks.', + description: + 'Processor used for the async task. Higher-end processors for longer-running tasks.', displayOptions: { show: { resource: ['task'], @@ -310,6 +319,16 @@ export class Parallel implements INodeType { options: TASK_PROCESSOR_OPTIONS, default: 'pro', }, + { + displayName: 'Maximum Wait (Minutes)', + name: 'syncWaitMinutes', + type: 'number', + typeOptions: { minValue: 1, maxValue: 120 }, + default: 75, + displayOptions: { show: { resource: ['task'], operation: ['webEnrichment'] } }, + description: + 'Total time to wait for an existing Task result. The n8n execution limit may be shorter. Use Async for long tasks.', + }, { displayName: 'Webhook URL', name: 'webhookUrl', @@ -322,7 +341,8 @@ export class Parallel implements INodeType { }, default: '', placeholder: 'https://your-domain.com/webhooks/parallel', - description: 'Optional webhook URL to receive real-time notifications when the task completes', + description: + 'Optional webhook URL to receive real-time notifications when the task completes', }, { displayName: 'Additional Fields', @@ -432,7 +452,7 @@ export class Parallel implements INodeType { description: 'Fastest search for English and Japanese queries', }, ], - default: 'advanced', + default: 'basic', }, { displayName: 'Additional Fields', @@ -591,30 +611,38 @@ export class Parallel implements INodeType { chatResponseFormat: ['json'], }, }, - default: JSON.stringify({ - type: 'object', - properties: { - answer: { - type: 'string', - description: 'Direct factual answer to the user question based on web research. If unavailable, return \'Information not found\'.', - }, - key_findings: { - type: 'array', - items: { type: 'string' }, - description: 'List of 3-5 most important facts or insights related to the question. Return empty array if no findings.', - }, - confidence_level: { - type: 'string', - description: 'Confidence level of the answer as \'High\', \'Medium\', or \'Low\' based on source quality and consistency.', - }, - last_updated_date: { - type: 'string', - description: 'Most recent date of information found in YYYY-MM-DD format. If unavailable, return \'Unknown\'.', + default: JSON.stringify( + { + type: 'object', + properties: { + answer: { + type: 'string', + description: + "Direct factual answer to the user question based on web research. If unavailable, return 'Information not found'.", + }, + key_findings: { + type: 'array', + items: { type: 'string' }, + description: + 'List of 3-5 most important facts or insights related to the question. Return empty array if no findings.', + }, + confidence_level: { + type: 'string', + description: + "Confidence level of the answer as 'High', 'Medium', or 'Low' based on source quality and consistency.", + }, + last_updated_date: { + type: 'string', + description: + "Most recent date of information found in YYYY-MM-DD format. If unavailable, return 'Unknown'.", + }, }, + required: ['answer', 'key_findings', 'confidence_level', 'last_updated_date'], + additionalProperties: false, }, - required: ['answer', 'key_findings', 'confidence_level', 'last_updated_date'], - additionalProperties: false, - }, null, 2), + null, + 2, + ), description: 'JSON schema defining the structure of the expected response', }, { @@ -639,7 +667,7 @@ export class Parallel implements INodeType { }, default: '', placeholder: 'You are a helpful assistant that provides accurate information...', - description: 'Optional system prompt to define the AI\'s behavior and role', + description: "Optional system prompt to define the AI's behavior and role", }, ], }, @@ -677,7 +705,8 @@ export class Parallel implements INodeType { }, default: '', placeholder: 'Track funding announcements for AI startups', - description: 'What to monitor - natural language description of the events to track on the web', + description: + 'What to monitor - natural language description of the events to track on the web', }, { displayName: 'Task Run ID', @@ -782,7 +811,8 @@ export class Parallel implements INodeType { monitorOutputSchemaType: ['json'], }, }, - default: '{\n "type": "object",\n "properties": {\n "company_name": {\n "type": "string",\n "description": "Company name"\n },\n "event_summary": {\n "type": "string",\n "description": "Brief summary of the event"\n },\n "sentiment": {\n "type": "string",\n "description": "Sentiment: positive, negative, or neutral"\n }\n }\n}', + default: + '{\n "type": "object",\n "properties": {\n "company_name": {\n "type": "string",\n "description": "Company name"\n },\n "event_summary": {\n "type": "string",\n "description": "Brief summary of the event"\n },\n "sentiment": {\n "type": "string",\n "description": "Sentiment: positive, negative, or neutral"\n }\n }\n}', description: 'JSON schema defining the structure of monitor event outputs', }, { @@ -797,7 +827,8 @@ export class Parallel implements INodeType { }, default: '', placeholder: 'https://your-n8n-instance.com/webhook/parallel-monitor-event', - description: 'Webhook URL to receive notifications when events are detected. Use the URL from a Parallel Monitor Event Trigger node.', + description: + 'Webhook URL to receive notifications when events are detected. Use the URL from a Parallel Monitor Event Trigger node.', }, { displayName: 'Webhook Event Types', @@ -888,7 +919,14 @@ export class Parallel implements INodeType { displayOptions: { show: { resource: ['monitor'], - monitorOperation: ['getMonitor', 'updateMonitor', 'deleteMonitor', 'listMonitorEvents', 'getMonitorEventGroup', 'triggerMonitor'], + monitorOperation: [ + 'getMonitor', + 'updateMonitor', + 'deleteMonitor', + 'listMonitorEvents', + 'getMonitorEventGroup', + 'triggerMonitor', + ], }, }, default: '', @@ -1088,6 +1126,14 @@ export class Parallel implements INodeType { default: false, description: 'Whether to include executions that completed without a detected change', }, + { + displayName: 'Legacy Lookback Period', + name: 'lookbackPeriod', + type: 'string', + default: '', + description: + 'Unsupported by the current API. Remove this saved option and use pagination; older history may be unavailable.', + }, { displayName: 'Limit', name: 'limit', @@ -1171,9 +1217,13 @@ export class Parallel implements INodeType { result = await operations.triggerMonitor.execute(this, i); break; default: - throw new NodeOperationError(this.getNode(), `Unknown monitor operation: ${monitorOperation}`, { - itemIndex: i, - }); + throw new NodeOperationError( + this.getNode(), + `Unknown monitor operation: ${monitorOperation}`, + { + itemIndex: i, + }, + ); } } else { const operation = this.getNodeParameter('operation', i) as string; diff --git a/nodes/Parallel/actions/asyncWebEnrichment.operation.ts b/nodes/Parallel/actions/asyncWebEnrichment.operation.ts index 84d01ea..06067d1 100644 --- a/nodes/Parallel/actions/asyncWebEnrichment.operation.ts +++ b/nodes/Parallel/actions/asyncWebEnrichment.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { NodeOperationError } from 'n8n-workflow'; import { assertTaskOutputSchemaCompatibility } from '../contracts/requests'; import { parallelApiRequest } from '../transport/ParallelApi'; @@ -11,7 +7,8 @@ import { buildSourcePolicy, buildMetadata } from '../utils'; export const description: INodePropertyOptions = { name: 'Async Web Enrichment', value: 'asyncWebEnrichment', - description: 'Create a Task with the Parallel Task API and retrieve its Run ID for async retrieval', + description: + 'Create a Task with the Parallel Task API and retrieve its Run ID for async retrieval', action: 'Async Web Enrichment', }; @@ -72,7 +69,10 @@ export async function execute( type: 'text', }; } else if (outputSchemaType === 'json') { - const jsonSchemaString = executeFunctions.getNodeParameter('asyncOutputJsonSchema', itemIndex) as string; + const jsonSchemaString = executeFunctions.getNodeParameter( + 'asyncOutputJsonSchema', + itemIndex, + ) as string; try { const jsonSchema = JSON.parse(jsonSchemaString); taskSpec.output_schema = { @@ -134,7 +134,8 @@ export async function execute( if (webhookUrl) { result.webhook_configured = true; result.webhook_url = webhookUrl; - result.message = 'Task started successfully with webhook notifications. You will receive a webhook call when the task completes.'; + result.message = + 'Task started successfully with webhook notifications. You will receive a webhook call when the task completes.'; } else { result.webhook_configured = false; } diff --git a/nodes/Parallel/actions/createMonitor.operation.ts b/nodes/Parallel/actions/createMonitor.operation.ts index 4edcc31..c00a05c 100644 --- a/nodes/Parallel/actions/createMonitor.operation.ts +++ b/nodes/Parallel/actions/createMonitor.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { NodeOperationError } from 'n8n-workflow'; import { buildMonitorCreateRequest, type MonitorType } from '../contracts/requests'; import { parallelApiRequest } from '../transport/ParallelApi'; @@ -11,7 +7,8 @@ import { buildMetadata } from '../utils'; export const description: INodePropertyOptions = { name: 'Create Monitor', value: 'createMonitor', - description: 'Create a new web monitor that continuously tracks the web for changes on a schedule', + description: + 'Create a new web monitor that continuously tracks the web for changes on a schedule', action: 'Create a monitor', }; @@ -21,26 +18,49 @@ export async function execute( ): Promise { const query = executeFunctions.getNodeParameter('monitorQuery', itemIndex) as string; const cadence = executeFunctions.getNodeParameter('monitorCadence', itemIndex) as string; - const type = executeFunctions.getNodeParameter('monitorType', itemIndex, 'event_stream') as MonitorType; + const type = executeFunctions.getNodeParameter( + 'monitorType', + itemIndex, + 'event_stream', + ) as MonitorType; const taskRunId = executeFunctions.getNodeParameter('monitorTaskRunId', itemIndex, '') as string; - const processor = executeFunctions.getNodeParameter('monitorProcessor', itemIndex, 'lite') as string; + const processor = executeFunctions.getNodeParameter( + 'monitorProcessor', + itemIndex, + 'lite', + ) as string; // Webhook URL - const webhookUrl = executeFunctions.getNodeParameter('monitorWebhookUrl', itemIndex, '') as string; + const webhookUrl = executeFunctions.getNodeParameter( + 'monitorWebhookUrl', + itemIndex, + '', + ) as string; const eventTypes = webhookUrl ? (executeFunctions.getNodeParameter('monitorWebhookEventTypes', itemIndex, []) as string[]) : undefined; // Output schema - const outputSchemaType = executeFunctions.getNodeParameter('monitorOutputSchemaType', itemIndex, 'text') as string; + const outputSchemaType = executeFunctions.getNodeParameter( + 'monitorOutputSchemaType', + itemIndex, + 'text', + ) as string; let outputSchema: IDataObject | undefined; if (type === 'event_stream' && outputSchemaType === 'json') { - const jsonSchemaStr = executeFunctions.getNodeParameter('monitorOutputJsonSchema', itemIndex) as string; + const jsonSchemaStr = executeFunctions.getNodeParameter( + 'monitorOutputJsonSchema', + itemIndex, + ) as string; let jsonSchema: IDataObject; try { jsonSchema = typeof jsonSchemaStr === 'string' ? JSON.parse(jsonSchemaStr) : jsonSchemaStr; } catch { - throw new NodeOperationError(executeFunctions.getNode(), 'Invalid JSON schema for monitor output. Please provide valid JSON.', { itemIndex }); + throw new NodeOperationError( + executeFunctions.getNode(), + 'Invalid JSON schema for monitor output. Please provide valid JSON.', + { itemIndex }, + ); } outputSchema = { type: 'json', @@ -49,7 +69,11 @@ export async function execute( } // Metadata - const additionalFields = executeFunctions.getNodeParameter('monitorAdditionalFields', itemIndex, {}) as IDataObject; + const additionalFields = executeFunctions.getNodeParameter( + 'monitorAdditionalFields', + itemIndex, + {}, + ) as IDataObject; const metadata = buildMetadata(additionalFields); let body: IDataObject; try { diff --git a/nodes/Parallel/actions/deleteMonitor.operation.ts b/nodes/Parallel/actions/deleteMonitor.operation.ts index 273a807..4795061 100644 --- a/nodes/Parallel/actions/deleteMonitor.operation.ts +++ b/nodes/Parallel/actions/deleteMonitor.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { parallelApiRequest } from '../transport/ParallelApi'; export const description: INodePropertyOptions = { diff --git a/nodes/Parallel/actions/getMonitor.operation.ts b/nodes/Parallel/actions/getMonitor.operation.ts index 1d76395..5dbc3ac 100644 --- a/nodes/Parallel/actions/getMonitor.operation.ts +++ b/nodes/Parallel/actions/getMonitor.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { parallelApiRequest } from '../transport/ParallelApi'; export const description: INodePropertyOptions = { diff --git a/nodes/Parallel/actions/getMonitorEventGroup.operation.ts b/nodes/Parallel/actions/getMonitorEventGroup.operation.ts index 6ad0d4e..95bb0d1 100644 --- a/nodes/Parallel/actions/getMonitorEventGroup.operation.ts +++ b/nodes/Parallel/actions/getMonitorEventGroup.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { parallelApiRequest } from '../transport/ParallelApi'; export const description: INodePropertyOptions = { diff --git a/nodes/Parallel/actions/getTaskRun.operation.ts b/nodes/Parallel/actions/getTaskRun.operation.ts index 1779dd0..53f2aa0 100644 --- a/nodes/Parallel/actions/getTaskRun.operation.ts +++ b/nodes/Parallel/actions/getTaskRun.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { parallelApiRequest } from '../transport/ParallelApi'; export const description: INodePropertyOptions = { diff --git a/nodes/Parallel/actions/getTaskRunResult.operation.ts b/nodes/Parallel/actions/getTaskRunResult.operation.ts index d8a7dcd..74ce5ec 100644 --- a/nodes/Parallel/actions/getTaskRunResult.operation.ts +++ b/nodes/Parallel/actions/getTaskRunResult.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { parallelApiRequest } from '../transport/ParallelApi'; export const description: INodePropertyOptions = { diff --git a/nodes/Parallel/actions/listMonitorEvents.operation.ts b/nodes/Parallel/actions/listMonitorEvents.operation.ts index 396a966..113be6f 100644 --- a/nodes/Parallel/actions/listMonitorEvents.operation.ts +++ b/nodes/Parallel/actions/listMonitorEvents.operation.ts @@ -1,15 +1,11 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { parallelApiRequest } from '../transport/ParallelApi'; import { buildMonitorEventsQuery } from '../contracts/requests'; export const description: INodePropertyOptions = { name: 'List Monitor Events', value: 'listMonitorEvents', - description: 'List events detected by a monitor within a lookback period', + description: 'List events detected by a monitor with cursor pagination', action: 'List monitor events', }; @@ -18,7 +14,11 @@ export async function execute( itemIndex: number, ): Promise { const monitorId = executeFunctions.getNodeParameter('monitorId', itemIndex) as string; - const additionalFields = executeFunctions.getNodeParameter('monitorEventsAdditionalFields', itemIndex, {}) as IDataObject; + const additionalFields = executeFunctions.getNodeParameter( + 'monitorEventsAdditionalFields', + itemIndex, + {}, + ) as IDataObject; return await parallelApiRequest( executeFunctions, diff --git a/nodes/Parallel/actions/listMonitors.operation.ts b/nodes/Parallel/actions/listMonitors.operation.ts index 99382e6..e82801a 100644 --- a/nodes/Parallel/actions/listMonitors.operation.ts +++ b/nodes/Parallel/actions/listMonitors.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { parallelApiRequest } from '../transport/ParallelApi'; export const description: INodePropertyOptions = { @@ -16,7 +12,11 @@ export async function execute( executeFunctions: IExecuteFunctions, itemIndex: number, ): Promise { - const additionalFields = executeFunctions.getNodeParameter('listMonitorsAdditionalFields', itemIndex, {}) as IDataObject; + const additionalFields = executeFunctions.getNodeParameter( + 'listMonitorsAdditionalFields', + itemIndex, + {}, + ) as IDataObject; const query: IDataObject = {}; if (additionalFields.limit) { diff --git a/nodes/Parallel/actions/triggerMonitor.operation.ts b/nodes/Parallel/actions/triggerMonitor.operation.ts index cc0618a..f54bef0 100644 --- a/nodes/Parallel/actions/triggerMonitor.operation.ts +++ b/nodes/Parallel/actions/triggerMonitor.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { parallelApiRequest } from '../transport/ParallelApi'; export const description: INodePropertyOptions = { diff --git a/nodes/Parallel/actions/updateMonitor.operation.ts b/nodes/Parallel/actions/updateMonitor.operation.ts index 5beae7e..4437c49 100644 --- a/nodes/Parallel/actions/updateMonitor.operation.ts +++ b/nodes/Parallel/actions/updateMonitor.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { NodeOperationError } from 'n8n-workflow'; import { buildMonitorUpdateRequest } from '../contracts/requests'; import { parallelApiRequest } from '../transport/ParallelApi'; @@ -11,7 +7,7 @@ import { buildMetadata } from '../utils'; export const description: INodePropertyOptions = { name: 'Update Monitor', value: 'updateMonitor', - description: 'Update an existing monitor\'s query, cadence, webhook, or metadata', + description: "Update an existing monitor's query, cadence, webhook, or metadata", action: 'Update a monitor', }; @@ -20,7 +16,11 @@ export async function execute( itemIndex: number, ): Promise { const monitorId = executeFunctions.getNodeParameter('monitorId', itemIndex) as string; - const updateFields = executeFunctions.getNodeParameter('monitorUpdateFields', itemIndex, {}) as IDataObject; + const updateFields = executeFunctions.getNodeParameter( + 'monitorUpdateFields', + itemIndex, + {}, + ) as IDataObject; const metadata = buildMetadata(updateFields); let body: IDataObject; diff --git a/nodes/Parallel/actions/webChat.operation.ts b/nodes/Parallel/actions/webChat.operation.ts index 7a76615..d9cd0d7 100644 --- a/nodes/Parallel/actions/webChat.operation.ts +++ b/nodes/Parallel/actions/webChat.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { NodeOperationError } from 'n8n-workflow'; import { parallelApiRequest } from '../transport/ParallelApi'; @@ -19,8 +15,15 @@ export async function execute( ): Promise { const inputPrompt = executeFunctions.getNodeParameter('chatInputPrompt', itemIndex) as string; const model = executeFunctions.getNodeParameter('chatModel', itemIndex, 'speed') as string; - const responseFormat = executeFunctions.getNodeParameter('chatResponseFormat', itemIndex) as string; - const additionalOptions = executeFunctions.getNodeParameter('chatAdditionalOptions', itemIndex, {}) as IDataObject; + const responseFormat = executeFunctions.getNodeParameter( + 'chatResponseFormat', + itemIndex, + ) as string; + const additionalOptions = executeFunctions.getNodeParameter( + 'chatAdditionalOptions', + itemIndex, + {}, + ) as IDataObject; // Build messages array const messages: IDataObject[] = []; @@ -48,13 +51,19 @@ export async function execute( // Add response format if JSON is selected if (responseFormat === 'json') { - const jsonSchemaName = executeFunctions.getNodeParameter('chatJsonSchemaName', itemIndex) as string; + const jsonSchemaName = executeFunctions.getNodeParameter( + 'chatJsonSchemaName', + itemIndex, + ) as string; if (!jsonSchemaName.trim()) { throw new NodeOperationError(executeFunctions.getNode(), 'JSON schema name is required.', { itemIndex, }); } - const jsonSchemaString = executeFunctions.getNodeParameter('chatJsonSchema', itemIndex) as string; + const jsonSchemaString = executeFunctions.getNodeParameter( + 'chatJsonSchema', + itemIndex, + ) as string; try { const jsonSchema = JSON.parse(jsonSchemaString); diff --git a/nodes/Parallel/actions/webEnrichment.operation.ts b/nodes/Parallel/actions/webEnrichment.operation.ts index 51531a3..e8b3224 100644 --- a/nodes/Parallel/actions/webEnrichment.operation.ts +++ b/nodes/Parallel/actions/webEnrichment.operation.ts @@ -1,11 +1,8 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { NodeOperationError } from 'n8n-workflow'; -import { getErrorStatusCode, parallelApiRequest } from '../transport/ParallelApi'; -import { buildSourcePolicy, buildMetadata, wait, calculateBackoffDelay, isRetryableError } from '../utils'; +import { parallelApiRequest } from '../transport/ParallelApi'; +import { waitForTaskResult } from '../transport/waitForTaskResult'; +import { buildSourcePolicy, buildMetadata } from '../utils'; export const description: INodePropertyOptions = { name: 'Sync Web Enrichment', @@ -61,7 +58,10 @@ export async function execute( type: 'text', }; } else if (outputSchemaType === 'json') { - const jsonSchemaString = executeFunctions.getNodeParameter('syncOutputJsonSchema', itemIndex) as string; + const jsonSchemaString = executeFunctions.getNodeParameter( + 'syncOutputJsonSchema', + itemIndex, + ) as string; try { const jsonSchema = JSON.parse(jsonSchemaString); taskSpec.output_schema = { @@ -82,10 +82,17 @@ export async function execute( let textDescription = ''; if (inputType === 'json') { // Required description when input is JSON - textDescription = executeFunctions.getNodeParameter('textOutputDescriptionRequired', itemIndex) as string; + textDescription = executeFunctions.getNodeParameter( + 'textOutputDescriptionRequired', + itemIndex, + ) as string; } else { // Optional description when input is text - const optionalDescription = executeFunctions.getNodeParameter('textOutputDescription', itemIndex, '') as string; + const optionalDescription = executeFunctions.getNodeParameter( + 'textOutputDescription', + itemIndex, + '', + ) as string; if (optionalDescription) { textDescription = optionalDescription; } @@ -93,7 +100,11 @@ export async function execute( // If we have a text description, modify the task spec to include it if (textDescription) { - if (taskSpec.output_schema && typeof taskSpec.output_schema === 'object' && (taskSpec.output_schema as IDataObject).type === 'text') { + if ( + taskSpec.output_schema && + typeof taskSpec.output_schema === 'object' && + (taskSpec.output_schema as IDataObject).type === 'text' + ) { // Add description to existing text schema (taskSpec.output_schema as IDataObject).description = textDescription; } else { @@ -125,50 +136,19 @@ export async function execute( body.source_policy = sourcePolicy; } - // Create task run + const minutes = Number(executeFunctions.getNodeParameter('syncWaitMinutes', itemIndex, 75)); + if (!Number.isFinite(minutes) || minutes < 1 || minutes > 120) + throw new NodeOperationError( + executeFunctions.getNode(), + 'Maximum Wait must be between 1 and 120 minutes', + { itemIndex }, + ); const taskRun = await parallelApiRequest(executeFunctions, 'POST', '/v1/tasks/runs', body); - const runId = taskRun.run_id; - - // Poll for result with exponential backoff retry logic - const maxAttempts = 15; - const deadline = Date.now() + 30 * 60 * 1000; - let attempt = 0; - - while (attempt < maxAttempts) { - try { - const timeout = 240; - const result = await parallelApiRequest( - executeFunctions, - 'GET', - `/v1/tasks/runs/${runId}/result?timeout=${timeout}`, - ); - - return result; - } catch (error) { - attempt++; - const statusCode = getErrorStatusCode(error); - - // Handle retryable errors with exponential backoff - if (isRetryableError(statusCode) && attempt < maxAttempts && Date.now() < deadline) { - const delay = calculateBackoffDelay(attempt); - - // Wait before retrying - await wait(delay); - continue; - } - - // For non-retryable errors or if we've exceeded max attempts, throw - throw new NodeOperationError( - executeFunctions.getNode(), - error instanceof Error ? error : String(error), - { itemIndex }, - ); - } - } - - throw new NodeOperationError( - executeFunctions.getNode(), - `Task execution timed out after ${maxAttempts} attempts (approximately ${maxAttempts * 4} minutes)`, - { itemIndex }, - ); + if (typeof taskRun.run_id !== 'string' || !taskRun.run_id) + throw new NodeOperationError( + executeFunctions.getNode(), + 'Task creation returned no run ID. Check Parallel before retrying to avoid duplicate tasks', + { itemIndex }, + ); + return await waitForTaskResult(executeFunctions, taskRun.run_id, minutes, itemIndex); } diff --git a/nodes/Parallel/actions/webSearch.operation.ts b/nodes/Parallel/actions/webSearch.operation.ts index 363e233..8571b53 100644 --- a/nodes/Parallel/actions/webSearch.operation.ts +++ b/nodes/Parallel/actions/webSearch.operation.ts @@ -1,8 +1,4 @@ -import type { - IExecuteFunctions, - IDataObject, - INodePropertyOptions, -} from 'n8n-workflow'; +import type { IExecuteFunctions, IDataObject, INodePropertyOptions } from 'n8n-workflow'; import { NodeOperationError } from 'n8n-workflow'; import { buildSearchRequest } from '../contracts/requests'; import { parallelApiRequest } from '../transport/ParallelApi'; @@ -10,7 +6,8 @@ import { parallelApiRequest } from '../transport/ParallelApi'; export const description: INodePropertyOptions = { name: 'Web Search', value: 'webSearch', - description: 'Search the web with the Parallel Search API and retrieve a list of results with excerpts', + description: + 'Search the web with the Parallel Search API and retrieve a list of results with excerpts', action: 'Web Search', }; diff --git a/nodes/Parallel/contracts/requests.ts b/nodes/Parallel/contracts/requests.ts index 9b7beb5..ff530cc 100644 --- a/nodes/Parallel/contracts/requests.ts +++ b/nodes/Parallel/contracts/requests.ts @@ -170,6 +170,10 @@ export function buildMonitorUpdateRequest(input: MonitorUpdateInput): IDataObjec } export function buildMonitorEventsQuery(input: IDataObject): IDataObject { + if (input.lookbackPeriod) + throw new Error( + 'Saved lookbackPeriod is unsupported by the current Monitor API. Remove it and use cursor pagination; full historical coverage is not guaranteed.', + ); const query: IDataObject = {}; if (input.eventGroupId) query.event_group_id = input.eventGroupId; if (input.cursor) query.cursor = input.cursor; diff --git a/nodes/Parallel/transport/ParallelApi.ts b/nodes/Parallel/transport/ParallelApi.ts index 2f9bf84..dcc800e 100644 --- a/nodes/Parallel/transport/ParallelApi.ts +++ b/nodes/Parallel/transport/ParallelApi.ts @@ -27,6 +27,7 @@ function requestOptions( endpoint: string, body?: IDataObject, qs?: IDataObject, + timeout?: number, ): IHttpRequestOptions { const options: IHttpRequestOptions = { method, @@ -34,6 +35,7 @@ function requestOptions( headers: { 'Content-Type': 'application/json' }, json: true, }; + if (timeout !== undefined) options.timeout = timeout; if (body !== undefined) options.body = body; if (qs && Object.keys(qs).length > 0) options.qs = qs; return options; @@ -45,12 +47,13 @@ export async function parallelApiRequest( endpoint: string, body?: IDataObject, qs?: IDataObject, + timeout?: number, ): Promise { try { const response: unknown = await context.helpers.httpRequestWithAuthentication.call( context, 'parallelApi', - requestOptions(method, endpoint, body, qs), + requestOptions(method, endpoint, body, qs, timeout), ); return normalizeResponse(response); } catch (error) { @@ -71,7 +74,7 @@ export async function parallelApiRequestForWebhook( const response: unknown = await context.helpers.httpRequestWithAuthentication.call( context, 'parallelApi', - requestOptions(method, endpoint, body, qs), + requestOptions(method, endpoint, body, qs, 5000), ); return normalizeResponse(response); } catch (error) { diff --git a/nodes/Parallel/transport/waitForTaskResult.ts b/nodes/Parallel/transport/waitForTaskResult.ts new file mode 100644 index 0000000..45f9f2f --- /dev/null +++ b/nodes/Parallel/transport/waitForTaskResult.ts @@ -0,0 +1,49 @@ +import type { IDataObject, IExecuteFunctions } from 'n8n-workflow'; +import { NodeOperationError } from 'n8n-workflow'; +import { calculateBackoffDelay, isRetryableError, wait } from '../utils'; +import { getErrorStatusCode, parallelApiRequest } from './ParallelApi'; + +// Keep the remote long-poll timeout and the local HTTP timeout inside one budget. +export async function waitForTaskResult( + context: IExecuteFunctions, + runId: string, + minutes: number, + itemIndex: number, + timing = { now: Date.now, sleep: wait }, +): Promise { + const deadline = timing.now() + minutes * 60_000; + let attempt = 0; + while (timing.now() < deadline) { + const remaining = deadline - timing.now(); + const timeout = Math.min(240, Math.floor(remaining / 1000)); + if (timeout < 1) break; + try { + return await parallelApiRequest( + context, + 'GET', + `/v1/tasks/runs/${encodeURIComponent(runId)}/result`, + undefined, + { timeout }, + Math.min(remaining, (timeout + 1) * 1000), + ); + } catch (error) { + if (!isRetryableError(getErrorStatusCode(error))) { + throw new NodeOperationError( + context.getNode(), + `Could not retrieve Task ${runId}. Retrieve this existing run with Get Task Run Result; do not resubmit. ${error instanceof Error ? error.message : String(error)}`, + { itemIndex }, + ); + } + const delay = Math.min( + calculateBackoffDelay(++attempt), + Math.max(0, deadline - timing.now()), + ); + if (delay > 0) await timing.sleep(delay); + } + } + throw new NodeOperationError( + context.getNode(), + `Timed out waiting for Task ${runId}. The Task is still remote and has not been cancelled. Use Get Task Run Result to retrieve this run or use Async for future long tasks; do not resubmit.`, + { itemIndex }, + ); +} diff --git a/nodes/Parallel/utils/index.ts b/nodes/Parallel/utils/index.ts index ded15ff..6ba61ea 100644 --- a/nodes/Parallel/utils/index.ts +++ b/nodes/Parallel/utils/index.ts @@ -61,7 +61,11 @@ export function wait(ms: number): Promise { /** * Calculates exponential backoff delay with jitter */ -export function calculateBackoffDelay(attempt: number, baseDelay: number = 2000, maxDelay: number = 60000): number { +export function calculateBackoffDelay( + attempt: number, + baseDelay: number = 2000, + maxDelay: number = 60000, +): number { // Calculate exponential backoff delay (base 2 seconds, max 60 seconds) const delay = Math.min(baseDelay * Math.pow(2, attempt - 1), maxDelay); diff --git a/nodes/Parallel/webhooks/response.ts b/nodes/Parallel/webhooks/response.ts new file mode 100644 index 0000000..fc019c7 --- /dev/null +++ b/nodes/Parallel/webhooks/response.ts @@ -0,0 +1,32 @@ +import type { IWebhookFunctions, IWebhookResponseData } from 'n8n-workflow'; +import { getParallelWebhookErrorMessage } from './verify'; + +// noWebhookResponse means n8n must not send again, not that it should ignore the event. +export function respond( + context: IWebhookFunctions, + status: number, + message?: string, +): IWebhookResponseData { + context + .getResponseObject() + .status(status) + .json(message ? { error: message } : { received: true }); + return { noWebhookResponse: true }; +} + +export async function validateWebhook( + context: IWebhookFunctions, +): Promise { + if (!(context.getNodeParameter('validateSignatures') as boolean)) return; + const credentials = await context.getCredentials('parallelApi'); + const headers = context.getHeaderData(); + const message = getParallelWebhookErrorMessage({ + secret: credentials.webhookSecret as string | undefined, + webhookId: headers['webhook-id'] as string | undefined, + webhookTimestamp: headers['webhook-timestamp'] as string | undefined, + signatureHeader: headers['webhook-signature'] as string | undefined, + rawBody: context.getRequestObject().rawBody, + }); + if (message) return respond(context, 401, message); + return undefined; +} diff --git a/nodes/Parallel/webhooks/verify.ts b/nodes/Parallel/webhooks/verify.ts index 4a3170b..a5baf0d 100644 --- a/nodes/Parallel/webhooks/verify.ts +++ b/nodes/Parallel/webhooks/verify.ts @@ -33,12 +33,19 @@ export function verifyParallelWebhook(input: VerifyWebhookInput): WebhookVerific const prefix = Buffer.from(`${input.webhookId}.${input.webhookTimestamp}.`, 'utf8'); const payload = Buffer.concat([prefix, input.rawBody]); - const expected = createHmac('sha256', secret).update(payload).digest(); + // Parallel supports Standard Webhooks and legacy HMAC over the complete secret. + const expectedSignatures = [secret, Buffer.from(input.secret, 'utf8')].map((key) => + createHmac('sha256', key).update(payload).digest(), + ); for (const candidate of input.signatureHeader.trim().split(/\s+/)) { if (!candidate.startsWith('v1,')) continue; const provided = Buffer.from(candidate.slice(3), 'base64'); - if (provided.length === expected.length && timingSafeEqual(expected, provided)) { + if ( + expectedSignatures.some( + (expected) => provided.length === expected.length && timingSafeEqual(expected, provided), + ) + ) { return { valid: true }; } } diff --git a/nodes/ParallelMonitorTrigger/ParallelMonitorTrigger.node.ts b/nodes/ParallelMonitorTrigger/ParallelMonitorTrigger.node.ts index e07d27f..41b2dca 100644 --- a/nodes/ParallelMonitorTrigger/ParallelMonitorTrigger.node.ts +++ b/nodes/ParallelMonitorTrigger/ParallelMonitorTrigger.node.ts @@ -6,25 +6,9 @@ import type { IWebhookFunctions, IWebhookResponseData, } from 'n8n-workflow'; -import { NodeApiError, NodeConnectionTypes } from 'n8n-workflow'; +import { NodeConnectionTypes } from 'n8n-workflow'; import { parallelApiRequestForWebhook } from '../Parallel/transport/ParallelApi'; -import { getParallelWebhookErrorMessage } from '../Parallel/webhooks/verify'; - -async function validateWebhook(context: IWebhookFunctions): Promise { - if (!(context.getNodeParameter('validateSignatures') as boolean)) return; - const credentials = await context.getCredentials('parallelApi'); - const headers = context.getHeaderData(); - const message = getParallelWebhookErrorMessage({ - secret: credentials.webhookSecret as string | undefined, - webhookId: headers['webhook-id'] as string | undefined, - webhookTimestamp: headers['webhook-timestamp'] as string | undefined, - signatureHeader: headers['webhook-signature'] as string | undefined, - rawBody: context.getRequestObject().rawBody, - }); - if (message) { - throw new NodeApiError(context.getNode(), {}, { message }); - } -} +import { respond, validateWebhook } from '../Parallel/webhooks/response'; export class ParallelMonitorTrigger implements INodeType { // Parallel receives this URL in the Monitor create/update request; activation does not register a remote webhook. @@ -49,7 +33,8 @@ export class ParallelMonitorTrigger implements INodeType { group: ['trigger'], version: 1, subtitle: '={{$parameter["eventTypeFilter"].join(", ")}}', - description: 'Triggers when a Parallel Monitor detects events, completes an execution, or encounters an error', + description: + 'Triggers when a Parallel Monitor detects events, completes an execution, or encounters an error', defaults: { name: 'Parallel Monitor Event', }, @@ -75,7 +60,8 @@ export class ParallelMonitorTrigger implements INodeType { name: 'webhookUrl', type: 'notice', default: '', - description: 'Use the webhook URL that n8n provides for this trigger node when creating or updating your Parallel monitor', + description: + 'Use the webhook URL that n8n provides for this trigger node when creating or updating your Parallel monitor', }, { displayName: 'Event Types', @@ -108,12 +94,21 @@ export class ParallelMonitorTrigger implements INodeType { default: true, description: 'Whether to fetch the full event group details when an event is detected', }, + { + displayName: 'Retry on Fetch Failure', + name: 'retryOnFetchFailure', + type: 'boolean', + default: false, + description: + 'Whether to ask Parallel to retry instead of emitting event_group_error when fetching fails. Retries can deliver the same event more than once.', + }, { displayName: 'Validate Webhook Signatures', name: 'validateSignatures', type: 'boolean', default: true, - description: 'Whether to validate the exact request body using the configured webhook secret', + description: + 'Whether to validate the exact request body using the configured webhook secret', }, { displayName: 'Include Webhook Data', @@ -123,19 +118,20 @@ export class ParallelMonitorTrigger implements INodeType { description: 'Whether to include the complete webhook payload in the output', }, ], - usableAsTool: true, }; async webhook(this: IWebhookFunctions): Promise { - await validateWebhook(this); + const invalid = await validateWebhook(this); + if (invalid) return invalid; const payload = this.getBodyData() as IDataObject; const eventType = String(payload.type ?? ''); const filters = this.getNodeParameter('eventTypeFilter') as string[]; - if (!filters.includes(eventType)) return { noWebhookResponse: true }; + if (!filters.includes(eventType)) return respond(this, 200); const data = payload.data as IDataObject | undefined; - if (!data?.monitor_id) return { noWebhookResponse: true }; + if (!data?.monitor_id) return respond(this, 400, 'Missing monitor_id'); const output: IDataObject = { + webhook_id: this.getHeaderData()['webhook-id'], event_type: eventType, monitor_id: String(data.monitor_id), timestamp: payload.timestamp, @@ -143,20 +139,26 @@ export class ParallelMonitorTrigger implements INodeType { metadata: data.metadata ?? {}, }; const event = data.event as IDataObject | undefined; - if ( - eventType === 'monitor.event.detected' && - (this.getNodeParameter('fetchEventGroup') as boolean) && - event?.event_group_id - ) { + if (eventType === 'monitor.event.detected') { + if (!event?.event_group_id) return respond(this, 400, 'Missing event_group_id'); output.event_group_id = event.event_group_id; - output.event_group = await parallelApiRequestForWebhook( - this, - 'GET', - `/v1/monitors/${encodeURIComponent(String(data.monitor_id))}/events`, - undefined, - { event_group_id: event.event_group_id }, - ); + if (this.getNodeParameter('fetchEventGroup') as boolean) { + try { + output.event_group = await parallelApiRequestForWebhook( + this, + 'GET', + `/v1/monitors/${encodeURIComponent(String(data.monitor_id))}/events`, + undefined, + { event_group_id: event.event_group_id }, + ); + } catch (error) { + if (this.getNodeParameter('retryOnFetchFailure', false) as boolean) + return respond(this, 503, 'Monitor events are unavailable; retry this delivery'); + output.event_group_error = error instanceof Error ? error.message : String(error); + } + } } + if (this.getNodeParameter('includeWebhookData') as boolean) { output.webhook_data = payload; } diff --git a/nodes/ParallelTrigger/ParallelTrigger.node.ts b/nodes/ParallelTrigger/ParallelTrigger.node.ts index 7cee2ad..6e848fe 100644 --- a/nodes/ParallelTrigger/ParallelTrigger.node.ts +++ b/nodes/ParallelTrigger/ParallelTrigger.node.ts @@ -6,25 +6,9 @@ import type { IWebhookFunctions, IWebhookResponseData, } from 'n8n-workflow'; -import { NodeApiError, NodeConnectionTypes } from 'n8n-workflow'; +import { NodeConnectionTypes } from 'n8n-workflow'; import { parallelApiRequestForWebhook } from '../Parallel/transport/ParallelApi'; -import { getParallelWebhookErrorMessage } from '../Parallel/webhooks/verify'; - -async function validateWebhook(context: IWebhookFunctions): Promise { - if (!(context.getNodeParameter('validateSignatures') as boolean)) return; - const credentials = await context.getCredentials('parallelApi'); - const headers = context.getHeaderData(); - const message = getParallelWebhookErrorMessage({ - secret: credentials.webhookSecret as string | undefined, - webhookId: headers['webhook-id'] as string | undefined, - webhookTimestamp: headers['webhook-timestamp'] as string | undefined, - signatureHeader: headers['webhook-signature'] as string | undefined, - rawBody: context.getRequestObject().rawBody, - }); - if (message) { - throw new NodeApiError(context.getNode(), {}, { message }); - } -} +import { respond, validateWebhook } from '../Parallel/webhooks/response'; export class ParallelTrigger implements INodeType { // Parallel receives this URL in the Task creation request; activation does not register a remote webhook. @@ -75,14 +59,16 @@ export class ParallelTrigger implements INodeType { name: 'webhookUrl', type: 'notice', default: '', - description: 'Use the webhook URL that n8n provides for this trigger node when configuring your Parallel task webhook', + description: + 'Use the webhook URL that n8n provides for this trigger node when configuring your Parallel task webhook', }, { displayName: 'Validate Webhook Signatures', name: 'validateSignatures', type: 'boolean', default: true, - description: 'Whether to validate the exact request body using the configured webhook secret', + description: + 'Whether to validate the exact request body using the configured webhook secret', }, { displayName: 'Only Trigger on Successful Tasks', @@ -99,33 +85,41 @@ export class ParallelTrigger implements INodeType { description: 'Whether to include the complete webhook payload in the output', }, ], - usableAsTool: true, }; async webhook(this: IWebhookFunctions): Promise { - await validateWebhook(this); + const invalid = await validateWebhook(this); + if (invalid) return invalid; const payload = this.getBodyData() as IDataObject; - if (payload.type !== 'task_run.status') return { noWebhookResponse: true }; + if (payload.type !== 'task_run.status') return respond(this, 200); const data = payload.data as IDataObject | undefined; - if (!data?.run_id || !data.status) return { noWebhookResponse: true }; + if (!data?.run_id || !data.status) return respond(this, 400, 'Missing Task run_id or status'); const status = String(data.status); - if (!['completed', 'failed'].includes(status)) return { noWebhookResponse: true }; + if (!['completed', 'failed'].includes(status)) return respond(this, 200); if ((this.getNodeParameter('onlyCompleted') as boolean) && status !== 'completed') { - return { noWebhookResponse: true }; + return respond(this, 200); } const output: IDataObject = { + webhook_id: this.getHeaderData()['webhook-id'], run_id: String(data.run_id), status, event: data, }; if (status === 'completed') { - output.result = await parallelApiRequestForWebhook( - this, - 'GET', - `/v1/tasks/runs/${encodeURIComponent(String(data.run_id))}/result`, - ); + try { + output.result = await parallelApiRequestForWebhook( + this, + 'GET', + `/v1/tasks/runs/${encodeURIComponent(String(data.run_id))}/result`, + undefined, + { timeout: 4 }, + ); + } catch { + return respond(this, 503, 'Task result is unavailable; retry this delivery'); + } } + if (this.getNodeParameter('includeWebhookData') as boolean) { output.webhook_data = payload; } diff --git a/package.json b/package.json index c5ceb49..64564d1 100644 --- a/package.json +++ b/package.json @@ -27,15 +27,17 @@ "scripts": { "build": "n8n-node build", "dev": "n8n-node dev", - "format": "prettier --write", + "format": "prettier --write .", "lint": "n8n-node lint", "lintfix": "n8n-node lint --fix", "typecheck": "tsc --noEmit --incremental false", "test": "npm run build && node --test test/*.test.js", - "prerelease": "n8n-node prerelease", "check-package": "node scripts/check-package.mjs", - "release": "n8n-node release", - "prepublishOnly": "npm run prerelease && npm run check-package" + "release": "node scripts/release.mjs", + "prepublishOnly": "n8n-node prerelease && npm run check-package", + "format:check": "prettier --check .", + "validate": "npm run format:check && npm run lint && npm run typecheck && npm test && npm audit --omit=dev && npm run check-package", + "release:rehearse": "node scripts/release.mjs --dry-run" }, "files": [ "dist" diff --git a/scripts/check-package.mjs b/scripts/check-package.mjs index 8e376a4..df4f9dd 100644 --- a/scripts/check-package.mjs +++ b/scripts/check-package.mjs @@ -4,14 +4,22 @@ import { mkdtempSync, readFileSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { createRequire } from 'node:module'; +import { fileURLToPath } from 'node:url'; -const root = new URL('..', import.meta.url).pathname; +const root = fileURLToPath(new URL('..', import.meta.url)); const temporaryDirectory = mkdtempSync(join(tmpdir(), 'parallel-n8n-package-')); try { const packOutput = execFileSync( 'npm', - ['pack', '--json', '--ignore-scripts', '--pack-destination', temporaryDirectory], + [ + 'pack', + '--dry-run=false', + '--json', + '--ignore-scripts', + '--pack-destination', + temporaryDirectory, + ], { cwd: root, encoding: 'utf8' }, ); const [packed] = JSON.parse(packOutput); @@ -25,13 +33,33 @@ try { assert.ok(![...packagedFiles].some((path) => path.endsWith('.tsbuildinfo'))); assert.ok(![...packagedFiles].some((path) => path.startsWith('dist/dist/'))); - execFileSync('npm', ['init', '--yes'], { cwd: temporaryDirectory, stdio: 'ignore' }); - execFileSync('npm', ['install', '--ignore-scripts', '--no-audit', '--no-fund', packagePath], { + execFileSync('npm', ['init', '--yes', '--dry-run=false'], { cwd: temporaryDirectory, stdio: 'ignore', }); + execFileSync( + 'npm', + ['install', '--dry-run=false', '--ignore-scripts', '--no-audit', '--no-fund', packagePath], + { + cwd: temporaryDirectory, + stdio: 'ignore', + }, + ); const require = createRequire(join(temporaryDirectory, 'consumer.cjs')); - require(join(temporaryDirectory, 'node_modules', manifest.name)); + const installed = join(temporaryDirectory, 'node_modules', manifest.name); + require(installed); + for (const entrypoint of [...manifest.n8n.credentials, ...manifest.n8n.nodes]) { + const exports = require(join(installed, entrypoint)); + const constructors = Object.values(exports).filter((value) => typeof value === 'function'); + assert.ok(constructors.length > 0, `No exported class: ${entrypoint}`); + for (const Constructor of constructors) { + const instance = new Constructor(); + assert.ok( + instance.description?.name || instance.name, + `Missing node/credential identity: ${entrypoint}`, + ); + } + } console.log(`Validated ${packed.filename} (${packagedFiles.size} files)`); } finally { diff --git a/scripts/release.mjs b/scripts/release.mjs new file mode 100644 index 0000000..ecd8cf0 --- /dev/null +++ b/scripts/release.mjs @@ -0,0 +1,28 @@ +import { execFileSync } from 'node:child_process'; +import { readFileSync } from 'node:fs'; + +// Keep validation real during rehearsal; only the final registry operation is a dry run. +const dryRun = process.argv.includes('--dry-run'); +const manifest = JSON.parse(readFileSync(new URL('../package.json', import.meta.url), 'utf8')); +if (!dryRun) { + if (process.env.GITHUB_ACTIONS !== 'true') + throw new Error('Publish through the reviewed GitHub Actions tag workflow'); + if (process.env.GITHUB_REF !== `refs/tags/v${manifest.version}`) + throw new Error('Release tag must match package.json version'); + const sha = execFileSync('git', ['rev-parse', 'HEAD'], { encoding: 'utf8' }).trim(); + if (sha !== process.env.GITHUB_SHA) + throw new Error('Checkout must match the release event commit'); + execFileSync('git', ['merge-base', '--is-ancestor', sha, 'origin/main']); +} +execFileSync('npm', ['run', 'validate'], { + stdio: 'inherit', + env: { ...process.env, npm_config_dry_run: 'false', NPM_CONFIG_DRY_RUN: 'false' }, +}); +execFileSync( + 'npm', + ['publish', '--provenance', '--access', 'public', ...(dryRun ? ['--dry-run'] : [])], + { + stdio: 'inherit', + env: { ...process.env, RELEASE_MODE: 'true' }, + }, +); diff --git a/test/reliability.test.js b/test/reliability.test.js new file mode 100644 index 0000000..3651eb0 --- /dev/null +++ b/test/reliability.test.js @@ -0,0 +1,225 @@ +const assert = require('node:assert/strict'); +const test = require('node:test'); +const { createHmac } = require('node:crypto'); +const { NodeHelpers } = require('n8n-workflow'); +const { Parallel } = require('../dist/nodes/Parallel/Parallel.node'); +const { ParallelTrigger } = require('../dist/nodes/ParallelTrigger/ParallelTrigger.node'); +const { + ParallelMonitorTrigger, +} = require('../dist/nodes/ParallelMonitorTrigger/ParallelMonitorTrigger.node'); +const { buildMonitorEventsQuery } = require('../dist/nodes/Parallel/contracts/requests'); +const { verifyParallelWebhook } = require('../dist/nodes/Parallel/webhooks/verify'); + +function context(payload, parameters = {}, request = async () => ({})) { + const replies = []; + const headers = { 'webhook-id': 'delivery_1' }; + const ctx = { + getNode: () => ({ name: 'Parallel', typeVersion: 1, parameters: {} }), + getBodyData: () => payload, + getHeaderData: () => headers, + getNodeParameter: (name, fallback) => + ({ + validateSignatures: false, + onlyCompleted: true, + includeWebhookData: false, + fetchEventGroup: true, + retryOnFetchFailure: false, + eventTypeFilter: ['monitor.event.detected'], + ...parameters, + })[name] ?? fallback, + getResponseObject: () => ({ + status(code) { + replies.push(code); + return this; + }, + json(body) { + replies.push(body); + return this; + }, + }), + helpers: { + returnJsonArray: (items) => items.map((json) => ({ json })), + httpRequestWithAuthentication: request, + }, + }; + return { ctx, replies }; +} + +test('legacy full-secret signatures preserve exact-byte verification', () => { + const rawBody = Buffer.from('{"a": 1}\n'); + const secret = `whsec_${Buffer.from('legacy-key').toString('base64')}`; + const timestamp = 1700000000; + const signature = createHmac('sha256', secret) + .update(Buffer.concat([Buffer.from(`id.${timestamp}.`), rawBody])) + .digest('base64'); + const input = { + secret, + rawBody, + webhookId: 'id', + webhookTimestamp: String(timestamp), + nowSeconds: timestamp, + signatureHeader: `v1,${signature}`, + }; + assert.equal(verifyParallelWebhook(input).valid, true); + assert.equal(verifyParallelWebhook({ ...input, rawBody: Buffer.from('{"a":1}') }).valid, false); +}); + +test('saved omitted Search mode retains the basic tier', () => { + const values = NodeHelpers.getNodeParameters( + new Parallel().description.properties, + { resource: 'task', operation: 'webSearch' }, + true, + false, + { typeVersion: 1 }, + ); + assert.equal(values.searchProcessor, 'basic'); +}); + +test('legacy Monitor lookback is retained by parameter loading and fails explicitly', () => { + const values = NodeHelpers.getNodeParameters( + new Parallel().description.properties, + { + resource: 'monitor', + monitorOperation: 'listMonitorEvents', + monitorEventsAdditionalFields: { lookbackPeriod: '7d' }, + }, + true, + false, + { typeVersion: 1 }, + ); + assert.equal(values.monitorEventsAdditionalFields.lookbackPeriod, '7d'); + assert.throws( + () => buildMonitorEventsQuery(values.monitorEventsAdditionalFields), + /lookbackPeriod.*pagination/i, + ); + assert.deepEqual(buildMonitorEventsQuery({ cursor: 'next', limit: 20 }), { + cursor: 'next', + limit: 20, + }); +}); + +test('valid filtered Task and Monitor events are acknowledged without workflow execution', async () => { + for (const [Node, payload] of [ + [ParallelTrigger, { type: 'task_run.status', data: { run_id: 'r1', status: 'failed' } }], + [ParallelMonitorTrigger, { type: 'monitor.execution.completed', data: { monitor_id: 'm1' } }], + ]) { + const { ctx, replies } = context(payload); + const result = await Node.prototype.webhook.call(ctx); + assert.equal(result.workflowData, undefined); + assert.deepEqual(replies, [200, { received: true }]); + } +}); + +test('malformed matching events return HTTP 400 without workflow execution', async () => { + for (const [Node, type] of [ + [ParallelTrigger, 'task_run.status'], + [ParallelMonitorTrigger, 'monitor.event.detected'], + ]) { + const { ctx, replies } = context({ type, data: {} }); + const result = await Node.prototype.webhook.call(ctx); + assert.equal(result.workflowData, undefined); + assert.equal(replies[0], 400); + } +}); + +test('Monitor keeps published top-level fields on no-fetch, success and failure paths', async () => { + const payload = { + type: 'monitor.event.detected', + data: { monitor_id: 'm1', event: { event_group_id: 'eg1' } }, + }; + for (const fetch of [false, true]) { + const { ctx } = context(payload, { fetchEventGroup: fetch }, async (_credential, options) => { + assert.equal(options.timeout, 5000); + return { events: [{ event_id: 'e1' }] }; + }); + const result = await ParallelMonitorTrigger.prototype.webhook.call(ctx); + assert.equal(result.workflowData[0][0].json.event_group_id, 'eg1'); + assert.equal(result.workflowData[0][0].json.webhook_id, 'delivery_1'); + } + const { ctx } = context(payload, {}, async () => { + throw { statusCode: 503, message: 'temporary failure' }; + }); + const result = await ParallelMonitorTrigger.prototype.webhook.call(ctx); + assert.equal(result.workflowData[0][0].json.event_group_id, 'eg1'); + assert.match(result.workflowData[0][0].json.event_group_error, /Parallel API/); +}); + +test('Monitor retry mode and Task enrichment failures send 503 without output', async () => { + for (const [Node, payload] of [ + [ParallelTrigger, { type: 'task_run.status', data: { run_id: 'r1', status: 'completed' } }], + [ + ParallelMonitorTrigger, + { + type: 'monitor.event.detected', + data: { monitor_id: 'm1', event: { event_group_id: 'eg1' } }, + }, + ], + ]) { + const { ctx, replies } = context(payload, { retryOnFetchFailure: true }, async () => { + throw { statusCode: 503 }; + }); + const result = await Node.prototype.webhook.call(ctx); + assert.equal(result.workflowData, undefined); + assert.equal(replies[0], 503); + } +}); + +test('Task polling caps each request and backoff and preserves the run ID on expiry', async () => { + const { waitForTaskResult } = require('../dist/nodes/Parallel/transport/waitForTaskResult'); + let now = 0; + let calls = 0; + const ctx = { + getNode: () => ({ name: 'Task', typeVersion: 1, parameters: {} }), + helpers: { + httpRequestWithAuthentication: async (_credential, options) => { + calls++; + assert.ok(options.timeout <= 60000 - now); + assert.ok(options.qs.timeout * 1000 <= 60000 - now); + now += options.timeout; + throw { statusCode: 408, message: 'pending' }; + }, + }, + }; + await assert.rejects( + waitForTaskResult(ctx, 'existing/run', 1, 0, { + now: () => now, + sleep: async (ms) => { + assert.ok(ms <= 60000 - now); + now += ms; + }, + }), + /existing\/run.*Get Task Run Result/, + ); + assert.equal(calls, 1); + assert.equal(now, 60000); +}); + +test('Task polling returns successful results and never recreates a task', async () => { + const { waitForTaskResult } = require('../dist/nodes/Parallel/transport/waitForTaskResult'); + let now = 0; + let calls = 0; + const ctx = { + getNode: () => ({ name: 'Task', typeVersion: 1, parameters: {} }), + helpers: { + httpRequestWithAuthentication: async (_credential, options) => { + assert.equal(options.method, 'GET'); + assert.match(options.url, /existing%2Frun\/result$/); + if (++calls === 1) { + now += 1000; + throw { statusCode: 408 }; + } + return { output: 'complete' }; + }, + }, + }; + assert.deepEqual( + await waitForTaskResult(ctx, 'existing/run', 1, 0, { + now: () => now, + sleep: async (ms) => { + now += ms; + }, + }), + { output: 'complete' }, + ); + assert.equal(calls, 2); +}); diff --git a/test/webhooks.test.js b/test/webhooks.test.js index 60329e5..d51c4fd 100644 --- a/test/webhooks.test.js +++ b/test/webhooks.test.js @@ -96,6 +96,7 @@ test('webhook validation explains missing and invalid inputs', () => { test('completed Task events encode the run ID in the result request path', async () => { let requestUrl; const context = { + getHeaderData: () => ({ 'webhook-id': 'test-delivery' }), getBodyData: () => ({ type: 'task_run.status', data: { run_id: 'trun/a b', status: 'completed' }, @@ -122,6 +123,7 @@ test('completed Task events encode the run ID in the result request path', async test('failed Task events are emitted without fetching the result endpoint', async () => { let requestCount = 0; const context = { + getHeaderData: () => ({ 'webhook-id': 'test-delivery' }), getBodyData: () => ({ type: 'task_run.status', data: { run_id: 'trun_failed', status: 'failed', error: { message: 'boom' } },