diff --git a/CHANGELOG.md b/CHANGELOG.md index f986bba..cd3cab9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## Unreleased + +### Removed + +- Removed the Relay v1 adaptor (`@openfn/language-registry-relay`), its + credential schema, example job, tests and workspace dependencies. Workflows + using that local adaptor must be updated before upgrading. No Relay v2 + replacement is included. + ## 0.1.0 Beta First self-hosted Evidence and BREG agricultural-holdings pilot. diff --git a/README.md b/README.md index 0ece034..d648342 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,12 @@ The maintained pilot uses: [Image identities](deployment/images.lock.json), the npm lockfile, and bootstrap [tool checksums](deployment/tools.sha256) pin the delivered dependencies. +Relay v1 support has been removed. The `@openfn/language-registry-relay` +package, including its helpers, credential schema and example job, is no longer +available through `OPENFN_ADAPTORS_REPO`. Existing workflows using it must be +updated before adopting this checkout. This repository does not currently +provide a Relay v2 adaptor. + ## What the three workflows do | Workflow | Authoritative result | Response and boundary | diff --git a/package-lock.json b/package-lock.json index 632ec30..4f40804 100644 --- a/package-lock.json +++ b/package-lock.json @@ -337,10 +337,6 @@ "resolved": "packages/registry-notary", "link": true }, - "node_modules/@openfn/language-registry-relay": { - "resolved": "packages/registry-relay", - "link": true - }, "node_modules/@openfn/lexicon": { "version": "2.4.0", "resolved": "https://registry.npmjs.org/@openfn/lexicon/-/lexicon-2.4.0.tgz", @@ -3040,70 +3036,6 @@ "semver": "^7.7.4", "source-map": "^0.7.6" } - }, - "packages/registry-relay": { - "name": "@openfn/language-registry-relay", - "version": "0.1.0", - "license": "Apache-2.0", - "dependencies": { - "@openfn/language-common": "3.3.3" - }, - "devDependencies": { - "@openfn/compiler": "1.2.5", - "@openfn/runtime": "1.9.3" - }, - "engines": { - "node": ">=20.18.1" - } - }, - "packages/registry-relay/node_modules/@openfn/compiler": { - "version": "1.2.5", - "resolved": "https://registry.npmjs.org/@openfn/compiler/-/compiler-1.2.5.tgz", - "integrity": "sha512-X0JcfC0rPfD/RJwMGr3uNOp+YiixJ86V1JintarndCUNOUI+Not9wq6sLQSz51YCMXbMezMhbY+QZN0e7WyzNA==", - "dev": true, - "license": "ISC", - "dependencies": { - "@openfn/describe-package": "0.1.6", - "@openfn/lexicon": "^2.0.0", - "@openfn/logger": "1.1.2", - "acorn": "^8.16.0", - "ast-types": "^0.14.2", - "recast": "^0.21.5" - }, - "engines": { - "node": ">=16", - "pnpm": ">=7" - } - }, - "packages/registry-relay/node_modules/@openfn/language-common": { - "version": "3.3.3", - "resolved": "https://registry.npmjs.org/@openfn/language-common/-/language-common-3.3.3.tgz", - "integrity": "sha512-i5uaM0inXROpU25C2zXV6h4rzEq19oMSGmg939vu3JBD5FxopyWR6ZMXAEho5EiFv7QPIHGueTdQbCQg+V5fWg==", - "license": "LGPL-3.0-or-later", - "dependencies": { - "ajv": "^8.20.0", - "csv-parse": "^5.4.0", - "csvtojson": "^2.0.13", - "date-fns": "^2.25.0", - "http-status-codes": "^2.3.0", - "jsonpath-plus": "^10.3.0", - "lodash": "^4.18.1", - "undici": "^7.24.7" - } - }, - "packages/registry-relay/node_modules/@openfn/runtime": { - "version": "1.9.3", - "resolved": "https://registry.npmjs.org/@openfn/runtime/-/runtime-1.9.3.tgz", - "integrity": "sha512-m4HeHYdWONWHjfayhKne7CgTIGx6l+p7MANevVIbrZs4dmFJb+g+DcHcikfRvW7S9sA2xBFnXK7FyMl/QjD7AQ==", - "dev": true, - "license": "ISC", - "dependencies": { - "@openfn/logger": "1.1.2", - "fast-safe-stringify": "^2.1.1", - "json-stream-stringify": "^3.1.6", - "semver": "^7.7.4", - "source-map": "^0.7.6" - } } } } diff --git a/packages/registry-relay/README.md b/packages/registry-relay/README.md deleted file mode 100644 index 0eba3ef..0000000 --- a/packages/registry-relay/README.md +++ /dev/null @@ -1,166 +0,0 @@ -# OpenFn Registry Relay Adaptor - -OpenFn helpers for reading protected Registry Relay APIs from workflows. - -Use this package when a workflow is authorized to read registry rows, metadata, -relationships, or aggregate outputs directly. Use -`@openfn/language-registry-notary` when the workflow needs a trust decision or a -certified value claim. - -When this repository is used as `OPENFN_ADAPTORS_REPO`, this package is loaded -as: - -```text -@openfn/language-registry-relay@local -``` - -## Configure - -Create an OpenFn credential with: - -- `relay_base_url`: Registry Relay service base URL. -- `token`: bearer token or API key for the Relay caller credential. - -The adaptor sends credentials as `Authorization: Bearer `. It does not -send `x-api-key`. - -The examples below use the public Registry Stack lab at -`https://lab.registrystack.org`. The lab publishes current credential metadata -at `https://lab.registrystack.org/api/lab.json`; use `agri-row-reader` for row -reads, `agri-aggregate-reader` for aggregate reads, `agri-metadata` for dataset -discovery, and `agri-evidence-only` for evidence offering discovery. - -## Read One Record - -```js -execute( - getRecord({ - dataset: "agri_registry", - entity: "farmer", - id: dataValue("farmer_id"), - purpose: "https://demo.example.gov/purpose/nagdi/climate-smart-input-support", - fields: ["id", "district", "registration_status"], - as: "farmer", - redactDataPaths: ["farmer_id"], - }), - - fn((state) => { - const farmer = state.data.farmer.record; - - return { - ...state, - data: { - ...state.data, - decision_input: { - farmer_id: farmer.id, - district: farmer.district, - relay_request_id: state.data.farmer.request_id, - }, - }, - }; - }), -); -``` - -## List Records - -Collection reads require an explicit `limit` and at least one filter unless -`allowUnfiltered: true` is set. - -```js -execute( - listRecords({ - dataset: "agri_registry", - entity: "farmer", - purpose: "https://demo.example.gov/purpose/nagdi/climate-smart-input-support", - filters: { - district: "north", - "id.in": ["FARMER-1001", "FARMER-1002"], - }, - fields: ["id", "district", "registration_status"], - limit: 50, - as: "farmers", - }), -); -``` - -## Query An Aggregate - -```js -execute( - queryAggregate({ - dataset: "agri_registry", - aggregate: "voucher_opportunities_by_district_crop_risk_input", - purpose: "https://demo.example.gov/purpose/nagdi/program-monitoring", - dimensions: ["district_code"], - measures: ["eligible_opportunity_count"], - filters: { season: ["2026A"] }, - maxRows: 100, - as: "district_summary", - }), - - fn((state) => { - const observations = state.data.district_summary.observations; - - return { - ...state, - data: { - ...state.data, - north_voucher_opportunities: - observations.find((row) => row.district_code === "north")?.eligible_opportunity_count ?? 0, - }, - }; - }), -); -``` - -## Discovery - -```js -execute( - discoverDatasets({ as: "catalog" }), - getEntitySchema({ - dataset: "agri_registry", - entity: "farmer", - as: "farmer_schema", - }), - listEvidenceOfferings({ as: "evidence_offerings" }), -); -``` - -## Result Branches - -Every helper writes its result under `state.data[as]`. If `as` is omitted, the -default names are `record`, `records`, `relationship`, `aggregate`, `datasets`, -`entity_schema`, or `evidence_offerings`. - -Common branches: - -- `succeeded` -- `not_modified` -- `not_found` -- `auth_failed` -- `forbidden` -- `filter_required` -- `cursor_invalid` -- `retryable_infrastructure` -- `failed` - -Problem Details are reduced to safe fields: `code`, `status`, `title`, and -`retryable`. The adaptor does not expose Problem Details `detail`. - -## Guardrails - -- Row, relationship, and aggregate helpers require `purpose`. -- `listRecords` requires `limit` and filters unless `allowUnfiltered: true`. -- Query values support OpenFn references such as `dataValue("farmer_id")`, plus - `{ valueFrom: "farmer_id" }` for simple path-based lookup. -- `X-Request-Id` uses `state.data.request_id` when present. -- `traceparent` is forwarded when `state.data.traceparent` is present. -- `ETag`, `Retry-After`, request id, and pagination cursors are preserved. -- Credentials, raw request material, and `configuration` are removed from final - state. - -Relay is a protected consultation API. It can publish evidence offerings, but it -does not evaluate trust decisions. Use the Registry Notary adaptor for claim -evaluation and certified value claims. diff --git a/packages/registry-relay/configuration-schema.json b/packages/registry-relay/configuration-schema.json deleted file mode 100644 index ca4c058..0000000 --- a/packages/registry-relay/configuration-schema.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "$schema": "http://json-schema.org/draft-07/schema#", - "type": "object", - "additionalProperties": true, - "required": [ - "relay_base_url", - "token" - ], - "properties": { - "relay_base_url": { - "title": "Registry Relay URL", - "type": "string", - "description": "Base URL for the Registry Relay service.", - "format": "uri", - "minLength": 1, - "examples": [ - "https://agriculture-relay.lab.registrystack.org" - ] - }, - "token": { - "title": "Bearer Token", - "type": "string", - "description": "Bearer token or API key for the Registry Relay caller credential.", - "writeOnly": true, - "minLength": 1 - } - } -} diff --git a/packages/registry-relay/jobs/read-record.js b/packages/registry-relay/jobs/read-record.js deleted file mode 100644 index d94fccb..0000000 --- a/packages/registry-relay/jobs/read-record.js +++ /dev/null @@ -1,32 +0,0 @@ -// SPDX-License-Identifier: Apache-2.0 - -import { dataValue, execute, fn } from "@openfn/language-common"; -import { getRecord } from "../src/index.js"; - -execute( - getRecord({ - dataset: "agri_registry", - entity: "farmer", - id: dataValue("farmer_id"), - purpose: "https://demo.example.gov/purpose/nagdi/climate-smart-input-support", - fields: ["id", "district", "registration_status"], - as: "farmer", - redactDataPaths: ["farmer_id"], - }), - - fn((state) => { - const farmer = state.data.farmer.record; - - return { - ...state, - data: { - ...state.data, - decision_input: { - farmer_id: farmer.id, - district: farmer.district, - relay_request_id: state.data.farmer.request_id, - }, - }, - }; - }), -); diff --git a/packages/registry-relay/package.json b/packages/registry-relay/package.json deleted file mode 100644 index d1eedbb..0000000 --- a/packages/registry-relay/package.json +++ /dev/null @@ -1,32 +0,0 @@ -{ - "name": "@openfn/language-registry-relay", - "version": "0.1.0", - "description": "OpenFn language helpers for Registry Relay.", - "license": "Apache-2.0", - "type": "module", - "main": "./src/index.js", - "exports": { - ".": "./src/index.js" - }, - "files": [ - "src", - "jobs", - "configuration-schema.json", - "README.md" - ], - "scripts": { - "test": "node --experimental-vm-modules --no-warnings --test test/relay.test.js", - "check": "node --check src/index.js && node --check jobs/read-record.js && npm test", - "pack:dry-run": "npm pack --dry-run" - }, - "engines": { - "node": ">=20.18.1" - }, - "dependencies": { - "@openfn/language-common": "3.3.3" - }, - "devDependencies": { - "@openfn/compiler": "1.2.5", - "@openfn/runtime": "1.9.3" - } -} diff --git a/packages/registry-relay/src/index.js b/packages/registry-relay/src/index.js deleted file mode 100644 index 70deccd..0000000 --- a/packages/registry-relay/src/index.js +++ /dev/null @@ -1,661 +0,0 @@ -// SPDX-License-Identifier: Apache-2.0 - -import { randomUUID } from "node:crypto"; - -const JSON_CONTENT_TYPES = new Set([ - "application/json", - "application/problem+json", - "application/vnd.sdmx.data+json", -]); -const UNSAFE_PATH_PARTS = new Set(["__proto__", "prototype", "constructor"]); - -export class RelayCallerError extends Error { - constructor(message, options = {}) { - super(message); - this.name = "RelayCallerError"; - this.code = options.code ?? "relay_caller.error"; - } -} - -export function getRecord(options = {}) { - return async (state) => callRelay(state, recordRequest(state, options)); -} - -export function listRecords(options = {}) { - return async (state) => callRelay(state, listRequest(state, options)); -} - -export function getRelationship(options = {}) { - return async (state) => callRelay(state, relationshipRequest(state, options)); -} - -export function queryAggregate(options = {}) { - return async (state) => callRelay(state, aggregateRequest(state, options)); -} - -export function discoverDatasets(options = {}) { - return async (state) => callRelay(state, simpleRequest(state, { - ...options, - kind: "datasets", - method: "GET", - path: "/v1/datasets", - purposeRequired: false, - })); -} - -export function getEntitySchema(options = {}) { - return async (state) => callRelay(state, simpleRequest(state, { - ...options, - kind: "entity_schema", - method: "GET", - path: `/v1/datasets/${pathPart(requiredResolvedString(state, options.dataset, "options.dataset"))}` - + `/entities/${pathPart(requiredResolvedString(state, options.entity, "options.entity"))}/schema`, - purposeRequired: false, - })); -} - -export function listEvidenceOfferings(options = {}) { - return async (state) => callRelay(state, simpleRequest(state, { - ...options, - kind: "evidence_offerings", - method: "GET", - path: "/metadata/evidence-offerings", - purposeRequired: false, - })); -} - -export function relayRequest(options = {}) { - return async (state) => callRelay(state, simpleRequest(state, { - ...options, - kind: options.kind ?? "relay_request", - method: options.method ?? "GET", - path: requiredResolvedString(state, options.path, "options.path"), - purposeRequired: options.purposeRequired ?? false, - })); -} - -export async function callRelay(state, request) { - const fetchImpl = request.fetch ?? globalThis.fetch; - if (typeof fetchImpl !== "function") { - throw new RelayCallerError("fetch is required to call Registry Relay", { - code: "fetch.required", - }); - } - - let response; - try { - response = await fetchImpl(request.url, request.init); - } catch (_error) { - return finish(state, request, { - branch: "retryable_infrastructure", - problem: { - code: "transport.error", - status: 0, - title: "Registry Relay request failed", - retryable: true, - }, - }); - } - - const headers = headersObject(response.headers); - const body = await readResponseBody(response, request); - const status = Number(response.status) || 0; - - if ((status >= 200 && status < 300) || status === 304) { - return finish(state, request, successPayload(request, status, headers, body)); - } - - return finish(state, request, problemPayload(request, status, headers, body)); -} - -export function buildRelayRequest(state, options = {}) { - return simpleRequest(state, options); -} - -export function redactRelayProblem(response) { - const body = response?.body ?? response?.data ?? response; - if (!body || typeof body !== "object" || Array.isArray(body)) { - return {}; - } - const { detail: _detail, ...safe } = body; - return safe; -} - -function recordRequest(state, options) { - const dataset = requiredResolvedString(state, options.dataset, "options.dataset"); - const entity = requiredResolvedString(state, options.entity, "options.entity"); - const id = requiredResolvedString(state, options.id ?? valueFromReference(state, options.idFrom), "options.id"); - return simpleRequest(state, { - ...options, - kind: "record", - method: "GET", - path: `/v1/datasets/${pathPart(dataset)}/entities/${pathPart(entity)}/records/${pathPart(id)}`, - query: { - fields: commaList(resolveInputValue(state, options.fields)), - expand: commaList(resolveInputValue(state, options.expand)), - }, - purposeRequired: true, - }); -} - -function listRequest(state, options) { - const dataset = requiredResolvedString(state, options.dataset, "options.dataset"); - const entity = requiredResolvedString(state, options.entity, "options.entity"); - const filters = compactObject(resolveInputValue(state, options.filters ?? {})); - if (!options.allowUnfiltered && Object.keys(filters).length === 0) { - throw new RelayCallerError("listRecords requires filters or allowUnfiltered: true", { - code: "filters.required", - }); - } - const limit = resolveInputValue(state, options.limit); - if (!Number.isSafeInteger(limit) || limit < 1) { - throw new RelayCallerError("listRecords requires a positive integer limit", { - code: "limit.required", - }); - } - return simpleRequest(state, { - ...options, - kind: "records", - method: "GET", - path: `/v1/datasets/${pathPart(dataset)}/entities/${pathPart(entity)}/records`, - query: { - limit, - cursor: resolveInputValue(state, options.cursor), - fields: commaList(resolveInputValue(state, options.fields)), - expand: commaList(resolveInputValue(state, options.expand)), - ...filters, - }, - purposeRequired: true, - }); -} - -function relationshipRequest(state, options) { - const dataset = requiredResolvedString(state, options.dataset, "options.dataset"); - const entity = requiredResolvedString(state, options.entity, "options.entity"); - const id = requiredResolvedString(state, options.id ?? valueFromReference(state, options.idFrom), "options.id"); - const relationship = requiredResolvedString(state, options.relationship, "options.relationship"); - const limit = resolveInputValue(state, options.limit); - if (limit !== undefined && (!Number.isSafeInteger(limit) || limit < 1)) { - throw new RelayCallerError("getRelationship limit must be a positive integer", { - code: "limit.invalid", - }); - } - return simpleRequest(state, { - ...options, - kind: "relationship", - method: "GET", - path: `/v1/datasets/${pathPart(dataset)}/entities/${pathPart(entity)}/records/${pathPart(id)}` - + `/relationships/${pathPart(relationship)}`, - query: { - limit, - cursor: resolveInputValue(state, options.cursor), - }, - purposeRequired: true, - }); -} - -function aggregateRequest(state, options) { - const dataset = requiredResolvedString(state, options.dataset, "options.dataset"); - const aggregate = requiredResolvedString(state, options.aggregate, "options.aggregate"); - const format = resolveInputValue(state, options.format) ?? "json"; - if (!["json", "csv", "sdmx-json"].includes(format)) { - throw new RelayCallerError("queryAggregate format must be json, csv, or sdmx-json", { - code: "format.invalid", - }); - } - const body = compactObject(resolveInputValue(state, options.body ?? { - filters: options.filters, - format, - group_by: options.groupBy ?? options.dimensions, - max_rows: options.maxRows, - measures: options.measures, - temporal: options.temporal, - })); - return simpleRequest(state, { - ...options, - kind: "aggregate", - method: "POST", - path: `/v1/datasets/${pathPart(dataset)}/aggregates/${pathPart(aggregate)}/query`, - query: { f: format }, - body, - purposeRequired: true, - responseFormat: format, - }); -} - -function simpleRequest(state, options) { - const configuration = configurationObject(state); - const data = dataObject(state); - const baseUrl = trimTrailingSlash(requireString(configuration.relay_base_url, "configuration.relay_base_url")); - const token = requireString(configuration.relay_token ?? configuration.token, "configuration.token"); - const method = requiredResolvedString(state, options.method ?? "GET", "options.method").toUpperCase(); - const path = normalizedPath(requiredResolvedString(state, options.path, "options.path")); - const purpose = resolveInputValue(state, options.purpose); - if (options.purposeRequired !== false) { - requireString(purpose, "options.purpose"); - } - const requestId = stringOrUndefined(data.request_id) - ?? stringOrUndefined(resolveInputValue(state, options.requestId)) - ?? randomUUID(); - const query = compactObject(resolveInputValue(state, options.query ?? {})); - const url = new URL(`${baseUrl}${path}`); - for (const [key, value] of Object.entries(query)) { - appendQuery(url, key, value); - } - const responseFormat = options.responseFormat ?? "json"; - const accept = resolveInputValue(state, options.accept); - - const headers = compactObject({ - Authorization: `Bearer ${token}`, - Accept: accept ?? acceptHeader(responseFormat), - "Content-Type": method === "POST" || method === "PUT" || method === "PATCH" ? "application/json" : undefined, - "Data-Purpose": purpose, - "If-None-Match": resolveInputValue(state, options.ifNoneMatch), - "X-Request-Id": requestId, - traceparent: stringOrUndefined(data.traceparent), - }); - const body = resolveInputValue(state, options.body); - - return { - kind: options.kind ?? "relay_request", - as: stringOrUndefined(options.as) ?? defaultResultName(options.kind), - url: url.toString(), - init: { - method, - headers, - ...(body === undefined ? {} : { body: JSON.stringify(body) }), - }, - fetch: options.fetch, - responseFormat, - purpose, - requestId, - redactDataPaths: redactDataPaths(options), - }; -} - -function successPayload(request, status, headers, body) { - const base = { - branch: status === 304 ? "not_modified" : "succeeded", - request_id: responseHeader(headers, "x-request-id") ?? request.requestId, - purpose: request.purpose, - etag: responseHeader(headers, "etag"), - }; - if (status === 304) { - return base; - } - switch (request.kind) { - case "record": - return { ...base, record: body }; - case "records": - case "relationship": - return { - ...base, - records: Array.isArray(body?.data) ? body.data : body, - pagination: body?.pagination, - }; - case "aggregate": - return { - ...base, - aggregate: body, - observations: Array.isArray(body?.observations) ? body.observations : undefined, - completeness: body?.completeness, - disclosure_control: body?.disclosure_control, - }; - case "datasets": - return { - ...base, - datasets: Array.isArray(body?.data) ? body.data : body, - }; - case "evidence_offerings": - return { - ...base, - offerings: Array.isArray(body?.evidence_offerings) ? body.evidence_offerings : body, - }; - case "entity_schema": - return { ...base, schema: body }; - default: - return { ...base, body }; - } -} - -function problemPayload(request, status, headers, body) { - const code = stringOrUndefined(body?.code) ?? statusToFallbackCode(status); - const branch = problemBranch(code, status); - const retryAfter = retryAfterSeconds(headers); - return { - branch, - request_id: responseHeader(headers, "x-request-id") ?? stringOrUndefined(body?.request_id) ?? request.requestId, - purpose: request.purpose, - ...(retryAfter !== undefined ? { retry_after_seconds: retryAfter } : {}), - problem: { - code, - status, - title: stringOrUndefined(body?.title), - retryable: branch === "retryable_infrastructure", - }, - }; -} - -function finish(state, request, result) { - const data = { ...dataObject(state) }; - delete data.relay_request; - delete data.relay_context; - for (const path of request.redactDataPaths) { - deleteDataPath(data, path); - } - const { configuration: _configuration, response: _response, ...safeState } = state; - return { - ...safeState, - data: { - ...data, - [request.as]: compactObject(result), - }, - }; -} - -async function readResponseBody(response, request) { - const text = await response.text(); - if (text.length === 0) { - return undefined; - } - if (request.responseFormat === "csv") { - return text; - } - const contentType = responseHeader(headersObject(response.headers), "content-type")?.split(";")[0]?.trim(); - if (contentType && !JSON_CONTENT_TYPES.has(contentType) && !contentType.endsWith("+json")) { - return text; - } - try { - return JSON.parse(text); - } catch (_error) { - throw new RelayCallerError("Registry Relay response was not valid JSON", { - code: "response.invalid_json", - }); - } -} - -function problemBranch(code, status) { - if (status === 404) { - return "not_found"; - } - if (status === 401 || code?.startsWith("auth.")) { - return "auth_failed"; - } - if (status === 403) { - return "forbidden"; - } - if (code === "entity.filter_required") { - return "filter_required"; - } - if (code?.includes("cursor")) { - return "cursor_invalid"; - } - if (status === 429 || status === 503 || status >= 500) { - return "retryable_infrastructure"; - } - return "failed"; -} - -function statusToFallbackCode(status) { - if (status === 401) { - return "auth.failed"; - } - if (status === 403) { - return "auth.forbidden"; - } - if (status === 404) { - return "resource.not_found"; - } - if (status === 429) { - return "rate_limited"; - } - if (status >= 500) { - return "source.unavailable"; - } - return "request.failed"; -} - -function retryAfterSeconds(headers) { - const raw = responseHeader(headers, "retry-after"); - if (raw === undefined) { - return undefined; - } - const parsed = Number(raw); - if (Number.isSafeInteger(parsed) && parsed > 0) { - return parsed; - } - const retryAt = Date.parse(raw); - if (!Number.isFinite(retryAt)) { - return undefined; - } - const serverDate = Date.parse(responseHeader(headers, "date") ?? ""); - const referenceTime = Number.isFinite(serverDate) ? serverDate : Date.now(); - const seconds = Math.ceil((retryAt - referenceTime) / 1000); - return Number.isSafeInteger(seconds) && seconds > 0 ? seconds : undefined; -} - -function acceptHeader(format) { - if (format === "csv") { - return "text/csv"; - } - if (format === "sdmx-json") { - return "application/vnd.sdmx.data+json;version=2.1, application/json"; - } - return "application/json"; -} - -function appendQuery(url, key, value) { - if (value === undefined || value === null || value === "") { - return; - } - if (Array.isArray(value)) { - url.searchParams.set(key, value.map((item) => String(item)).join(",")); - return; - } - if (typeof value === "object") { - throw new RelayCallerError(`query parameter ${key} must be scalar or array`, { - code: "query.invalid", - }); - } - url.searchParams.set(key, String(value)); -} - -function commaList(value) { - if (Array.isArray(value)) { - return value.join(","); - } - return value; -} - -function compactObject(value) { - if (!value || typeof value !== "object" || Array.isArray(value)) { - return value; - } - const compacted = {}; - for (const [key, item] of Object.entries(value)) { - if (item !== undefined && item !== null && item !== "") { - compacted[key] = item; - } - } - return compacted; -} - -function resolveInputValue(state, value) { - if (typeof value === "function") { - return value(state); - } - if (Array.isArray(value)) { - return value.map((item) => resolveInputValue(state, item)); - } - if (value && typeof value === "object") { - if (typeof value.valueFrom === "string") { - return valueFromPath(dataObject(state), value.valueFrom); - } - const resolved = {}; - for (const [key, item] of Object.entries(value)) { - resolved[key] = resolveInputValue(state, item); - } - return resolved; - } - return value; -} - -function valueFromReference(state, path) { - return typeof path === "string" ? valueFromPath(dataObject(state), path) : undefined; -} - -function valueFromPath(data, path) { - const parts = safePathParts(path); - if (!parts || parts.length === 0) { - return undefined; - } - let current = data; - for (const part of parts) { - if (!current || typeof current !== "object" || Array.isArray(current) || !Object.hasOwn(current, part)) { - return undefined; - } - current = current[part]; - } - return current; -} - -function redactDataPaths(options) { - const paths = new Set(Array.isArray(options.redactDataPaths) ? options.redactDataPaths : []); - collectValueFromPaths(options, paths); - return [...paths].filter((path) => typeof path === "string" && path.length > 0); -} - -function collectValueFromPaths(value, paths) { - if (Array.isArray(value)) { - for (const item of value) { - collectValueFromPaths(item, paths); - } - return; - } - if (!value || typeof value !== "object") { - return; - } - if (typeof value.valueFrom === "string") { - paths.add(value.valueFrom); - } - if (typeof value.idFrom === "string") { - paths.add(value.idFrom); - } - for (const item of Object.values(value)) { - collectValueFromPaths(item, paths); - } -} - -function deleteDataPath(data, path) { - const parts = safePathParts(path); - if (!parts || parts.length === 0) { - return; - } - if (parts.length === 1) { - delete data[parts[0]]; - return; - } - let current = data; - for (let index = 0; index < parts.length - 1; index += 1) { - const part = parts[index]; - if (!current || typeof current !== "object" || Array.isArray(current) || !Object.hasOwn(current, part)) { - return; - } - const next = current[part]; - if (!next || typeof next !== "object" || Array.isArray(next)) { - return; - } - current[part] = { ...next }; - current = current[part]; - } - delete current[parts[parts.length - 1]]; -} - -function safePathParts(path) { - if (typeof path !== "string" || path.length === 0) { - return undefined; - } - const parts = path.split(".").filter((part) => part.length > 0); - if (parts.some((part) => UNSAFE_PATH_PARTS.has(part))) { - return undefined; - } - return parts; -} - -function dataObject(state) { - return state?.data && typeof state.data === "object" && !Array.isArray(state.data) ? state.data : {}; -} - -function configurationObject(state) { - return state?.configuration && typeof state.configuration === "object" && !Array.isArray(state.configuration) - ? state.configuration - : {}; -} - -function requireString(value, label) { - if (typeof value !== "string" || value.length === 0) { - throw new RelayCallerError(`${label} is required`, { code: "value.required" }); - } - return value; -} - -function requiredResolvedString(state, value, label) { - return requireString(String(resolveInputValue(state, value) ?? ""), label); -} - -function stringOrUndefined(value) { - return typeof value === "string" && value.length > 0 ? value : undefined; -} - -function trimTrailingSlash(value) { - return value.replace(/\/+$/, ""); -} - -function pathPart(value) { - return encodeURIComponent(value); -} - -function normalizedPath(path) { - if (/^https?:\/\//i.test(path)) { - throw new RelayCallerError("options.path must be relative", { code: "path.absolute" }); - } - return path.startsWith("/") ? path : `/${path}`; -} - -function headersObject(headers) { - if (!headers) { - return {}; - } - if (typeof headers.forEach === "function") { - const result = {}; - headers.forEach((value, key) => { - result[key] = value; - }); - return result; - } - return { ...headers }; -} - -function responseHeader(headers, name) { - if (!headers || typeof headers !== "object") { - return undefined; - } - const exact = headers[name]; - if (typeof exact === "string") { - return exact; - } - const lowerName = name.toLowerCase(); - for (const [key, value] of Object.entries(headers)) { - if (key.toLowerCase() === lowerName && typeof value === "string") { - return value; - } - } - return undefined; -} - -function defaultResultName(kind) { - if (typeof kind === "string" && kind.length > 0) { - return kind; - } - return "relay"; -} diff --git a/packages/registry-relay/test/relay.test.js b/packages/registry-relay/test/relay.test.js deleted file mode 100644 index 2822770..0000000 --- a/packages/registry-relay/test/relay.test.js +++ /dev/null @@ -1,399 +0,0 @@ -import assert from "node:assert/strict"; -import { existsSync, readFileSync } from "node:fs"; -import { resolve } from "node:path"; -import test from "node:test"; -import { fileURLToPath } from "node:url"; - -import compile from "@openfn/compiler"; -import run from "@openfn/runtime"; -import { - RelayCallerError, - buildRelayRequest, - discoverDatasets, - getRecord, - listRecords, - queryAggregate, - redactRelayProblem, -} from "../src/index.js"; - -const baseState = Object.freeze({ - data: { - request_id: "wf-req-1", - farmer_id: "FARMER-1001", - traceparent: "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00", - }, - configuration: { - relay_base_url: "https://relay.example", - token: "secret-token", - }, -}); -const packageRoot = fileURLToPath(new URL("..", import.meta.url)); -const languageCommonRoot = findDependencyRoot(packageRoot, "@openfn/language-common"); - -test("getRecord sends bearer auth, purpose, request id, traceparent, and projected fields", async () => { - const calls = []; - const state = await getRecord({ - dataset: "agri_registry", - entity: "farmer", - id: { valueFrom: "farmer_id" }, - purpose: "https://demo.example.gov/purpose/nagdi/climate-smart-input-support", - fields: ["id", "district"], - as: "farmer", - fetch: async (url, init) => { - calls.push({ url, init }); - return jsonResponse( - { id: "FARMER-1001", district: "north" }, - { headers: { etag: "\"abc\"", "x-request-id": "relay-req-1" } }, - ); - }, - })(baseState); - - assert.equal(calls.length, 1); - assert.equal( - calls[0].url, - "https://relay.example/v1/datasets/agri_registry/entities/farmer/records/FARMER-1001?fields=id%2Cdistrict", - ); - assert.equal(calls[0].init.method, "GET"); - assert.equal(calls[0].init.headers.Authorization, "Bearer secret-token"); - assert.equal(calls[0].init.headers["Data-Purpose"], "https://demo.example.gov/purpose/nagdi/climate-smart-input-support"); - assert.equal(calls[0].init.headers["X-Request-Id"], "wf-req-1"); - assert.equal(calls[0].init.headers.traceparent, baseState.data.traceparent); - assert.deepEqual(state.data.farmer, { - branch: "succeeded", - record: { id: "FARMER-1001", district: "north" }, - request_id: "relay-req-1", - purpose: "https://demo.example.gov/purpose/nagdi/climate-smart-input-support", - etag: "\"abc\"", - }); - assert.equal("farmer_id" in state.data, false); - assert.equal("configuration" in state, false); - assert.equal(JSON.stringify(state).includes("secret-token"), false); -}); - -test("getRecord requires a purpose for row reads", async () => { - await assert.rejects( - () => - getRecord({ - dataset: "agri_registry", - entity: "farmer", - id: "FARMER-1001", - fetch: async () => jsonResponse({}), - })(baseState), - (error) => error instanceof RelayCallerError && error.code === "value.required", - ); -}); - -test("listRecords requires explicit limit and filters by default", async () => { - await assert.rejects( - () => - listRecords({ - dataset: "agri_registry", - entity: "farmer", - purpose: "purpose", - limit: 50, - fetch: async () => jsonResponse({ data: [], pagination: { has_more: false, next_cursor: null } }), - })(baseState), - (error) => error instanceof RelayCallerError && error.code === "filters.required", - ); - - await assert.rejects( - () => - listRecords({ - dataset: "agri_registry", - entity: "farmer", - purpose: "purpose", - filters: { district: "north" }, - fetch: async () => jsonResponse({ data: [], pagination: { has_more: false, next_cursor: null } }), - })(baseState), - (error) => error instanceof RelayCallerError && error.code === "limit.required", - ); -}); - -test("listRecords encodes filters, pagination, fields, and expansions", async () => { - const calls = []; - const state = await listRecords({ - dataset: "agri_registry", - entity: "farmer", - purpose: "purpose", - filters: { - district: "north", - "id.in": ["FARMER-1001", "FARMER-1002"], - }, - fields: ["id", "district"], - expand: ["holding"], - limit: 2, - cursor: "cursor-1", - as: "farmers", - fetch: async (url, init) => { - calls.push({ url, init }); - return jsonResponse({ - data: [{ id: "FARMER-1001" }], - pagination: { has_more: true, next_cursor: "cursor-2" }, - }); - }, - })(baseState); - - const url = new URL(calls[0].url); - assert.equal(url.pathname, "/v1/datasets/agri_registry/entities/farmer/records"); - assert.equal(url.searchParams.get("district"), "north"); - assert.equal(url.searchParams.get("id.in"), "FARMER-1001,FARMER-1002"); - assert.equal(url.searchParams.get("fields"), "id,district"); - assert.equal(url.searchParams.get("expand"), "holding"); - assert.equal(url.searchParams.get("limit"), "2"); - assert.equal(url.searchParams.get("cursor"), "cursor-1"); - assert.deepEqual(state.data.farmers.records, [{ id: "FARMER-1001" }]); - assert.deepEqual(state.data.farmers.pagination, { has_more: true, next_cursor: "cursor-2" }); -}); - -test("queryAggregate sends Relay aggregate query body", async () => { - const calls = []; - const state = await queryAggregate({ - dataset: "agri_registry", - aggregate: "voucher_opportunities_by_district_crop_risk_input", - purpose: "purpose", - dimensions: ["district_code"], - measures: ["eligible_opportunity_count"], - filters: { season: ["2026A"] }, - maxRows: 100, - as: "district_summary", - fetch: async (url, init) => { - calls.push({ url, init }); - return jsonResponse({ - dataset_id: "agri_registry", - aggregate_id: "voucher_opportunities_by_district_crop_risk_input", - observations: [{ district_code: "north", eligible_opportunity_count: 12 }], - completeness: { complete: true, truncated: false }, - disclosure_control: { method: ["k-anonymity"] }, - freshness: { computed_at: "2026-06-01T00:00:00Z" }, - links: [], - structure: { dimensions: [], measures: [] }, - }); - }, - })(baseState); - - assert.equal(calls[0].url, "https://relay.example/v1/datasets/agri_registry/aggregates/voucher_opportunities_by_district_crop_risk_input/query?f=json"); - assert.equal(calls[0].init.method, "POST"); - assert.equal(calls[0].init.headers["Content-Type"], "application/json"); - assert.deepEqual(JSON.parse(calls[0].init.body), { - filters: { season: ["2026A"] }, - format: "json", - group_by: ["district_code"], - max_rows: 100, - measures: ["eligible_opportunity_count"], - }); - assert.deepEqual(state.data.district_summary.observations, [{ district_code: "north", eligible_opportunity_count: 12 }]); - assert.deepEqual(state.data.district_summary.completeness, { complete: true, truncated: false }); -}); - -test("discoverDatasets does not send Data-Purpose", async () => { - const calls = []; - const state = await discoverDatasets({ - fetch: async (url, init) => { - calls.push({ url, init }); - return jsonResponse({ data: [{ dataset_id: "agri_registry" }] }); - }, - })(baseState); - - assert.equal(calls[0].url, "https://relay.example/v1/datasets"); - assert.equal("Data-Purpose" in calls[0].init.headers, false); - assert.deepEqual(state.data.datasets.datasets, [{ dataset_id: "agri_registry" }]); -}); - -test("Problem Details are mapped without leaking detail", async () => { - const state = await getRecord({ - dataset: "agri_registry", - entity: "farmer", - id: "FARMER-1001", - purpose: "purpose", - fetch: async () => - jsonResponse( - { - type: "https://relay.example/problems/entity.filter_required", - title: "Filter required", - status: 400, - detail: "secret FARMER-1001", - code: "entity.filter_required", - request_id: "problem-req-1", - }, - { status: 400, headers: { "x-request-id": "problem-req-1" } }, - ), - })(baseState); - - assert.equal(state.data.record.branch, "filter_required"); - assert.deepEqual(state.data.record.problem, { - code: "entity.filter_required", - status: 400, - title: "Filter required", - retryable: false, - }); - assert.equal(JSON.stringify(state).includes("secret FARMER-1001"), false); -}); - -test("429 and 503 responses become retryable infrastructure branches", async () => { - for (const status of [429, 503]) { - const state = await getRecord({ - dataset: "agri_registry", - entity: "farmer", - id: "FARMER-1001", - purpose: "purpose", - fetch: async () => - jsonResponse( - { title: "Retry later", status, code: "rate_limited", request_id: "retry-req" }, - { status, headers: { "retry-after": "3" } }, - ), - })(baseState); - - assert.equal(state.data.record.branch, "retryable_infrastructure"); - assert.equal(state.data.record.problem.retryable, true); - assert.equal(state.data.record.retry_after_seconds, 3); - } -}); - -test("transport errors are redacted retryable infrastructure branches", async () => { - const state = await getRecord({ - dataset: "agri_registry", - entity: "farmer", - id: "FARMER-1001", - purpose: "purpose", - fetch: async () => { - throw new Error("socket failed for secret FARMER-1001"); - }, - })(baseState); - - assert.equal(state.data.record.branch, "retryable_infrastructure"); - assert.equal(state.data.record.problem.code, "transport.error"); - assert.equal(JSON.stringify(state).includes("secret FARMER-1001"), false); -}); - -test("304 responses return not_modified with response validators", async () => { - const state = await getRecord({ - dataset: "agri_registry", - entity: "farmer", - id: "FARMER-1001", - purpose: "purpose", - ifNoneMatch: "\"abc\"", - fetch: async (_url, init) => { - assert.equal(init.headers["If-None-Match"], "\"abc\""); - return { - status: 304, - headers: { etag: "\"abc\"" }, - text: async () => "", - }; - }, - })(baseState); - - assert.equal(state.data.record.branch, "not_modified"); - assert.equal(state.data.record.etag, "\"abc\""); -}); - -test("buildRelayRequest rejects absolute escape-hatch paths", () => { - assert.throws( - () => - buildRelayRequest(baseState, { - path: "https://evil.example/v1/datasets", - }), - (error) => error instanceof RelayCallerError && error.code === "path.absolute", - ); -}); - -test("redactRelayProblem drops detail", () => { - assert.deepEqual( - redactRelayProblem({ - title: "Problem", - status: 400, - detail: "secret", - code: "request.failed", - request_id: "req-1", - }), - { - title: "Problem", - status: 400, - code: "request.failed", - request_id: "req-1", - }, - ); -}); - -test("template imports helper operations and avoids generic HTTP helpers", () => { - const template = readFileSync(new URL("../jobs/read-record.js", import.meta.url), "utf8"); - - assert.match(template, /getRecord/); - assert.match(template, /import\s+\{\s*dataValue,\s*execute,\s*fn\s*\}\s+from\s+["']@openfn\/language-common["']/); - assert.match(template, /from\s+["']\.\.\/src\/index\.js["']/); - assert.doesNotMatch(template, /from\s+["']@openfn\/language-http["']/); - assert.doesNotMatch(template, /Authorization/); -}); - -test("compiled OpenFn template runs through the OpenFn runtime and calls Relay once", async () => { - const template = readFileSync(new URL("../jobs/read-record.js", import.meta.url), "utf8"); - const { code } = compile(template); - const calls = []; - const originalFetch = globalThis.fetch; - globalThis.fetch = async (url, init) => { - calls.push({ url, init }); - return jsonResponse( - { id: "FARMER-1001", district: "north", registration_status: "active" }, - { headers: { "x-request-id": "relay-runtime-req" } }, - ); - }; - - try { - const result = await run( - { - workflow: { - steps: [{ id: "read-record", expression: code }], - start: "read-record", - }, - options: { start: "read-record" }, - }, - baseState, - { - linker: { - modules: { - "@openfn/language-common": { path: languageCommonRoot }, - "../src/index.js": { path: packageRoot }, - }, - cacheKey: `openfn-relay-test-${process.pid}-${Date.now()}`, - }, - statePropsToRemove: [], - }, - ); - - assert.equal(result.errors, undefined); - assert.equal(calls.length, 1); - assert.equal(calls[0].url, "https://relay.example/v1/datasets/agri_registry/entities/farmer/records/FARMER-1001?fields=id%2Cdistrict%2Cregistration_status"); - assert.equal(calls[0].init.headers.Authorization, "Bearer secret-token"); - assert.equal(result.data.farmer.branch, "succeeded"); - assert.equal(result.data.farmer.request_id, "relay-runtime-req"); - assert.equal(result.data.decision_input.district, "north"); - assert.equal("farmer_id" in result.data, false); - assert.equal("configuration" in result, false); - } finally { - globalThis.fetch = originalFetch; - } -}); - -function findDependencyRoot(start, packageName) { - let current = resolve(start); - while (true) { - const candidate = resolve(current, "node_modules", packageName); - if (existsSync(resolve(candidate, "package.json"))) { - return candidate; - } - const parent = resolve(current, ".."); - if (parent === current) { - throw new Error(`dependency not found: ${packageName}`); - } - current = parent; - } -} - -function jsonResponse(body, options = {}) { - return new Response(JSON.stringify(body), { - status: options.status ?? 200, - headers: { - "content-type": "application/json", - ...(options.headers ?? {}), - }, - }); -}