Node.js collector SDK

@seamward/collector observes third-party HTTP APIs, webhooks, queue operations, and scheduled feeds from a Node.js backend. It batches redacted evidence asynchronously and never sits on your critical request path. Collector failures do not throw into your application, while errors from your own handlers and clients are recorded and rethrown unchanged.

Requires Node.js 22 or later; the package is ESM-only. See Node collector setup for installation and a guided first integration.

createSeamwardCollector

import { createSeamwardCollector } from "@seamward/collector";
const seamward = createSeamwardCollector({  connectionKey: process.env.SEAMWARD_CONNECTION_KEY!,  ingestToken: process.env.SEAMWARD_INGEST_TOKEN!,});

Construction fails fast when the Connection key or ingest token is invalid, when hashKey is shorter than 16 characters, or when valueProfile is "full" without a hashKey. Key values are never echoed in errors. An unrecognised valueProfile is treated as "off".

Deployment context is resolved once at construction. Without explicit values, the collector reads SEAMWARD_SERVICE, SEAMWARD_RELEASE, and the first commit SHA found among SEAMWARD_COMMIT_SHA, VERCEL_GIT_COMMIT_SHA, RENDER_GIT_COMMIT, RAILWAY_GIT_COMMIT_SHA, and GITHUB_SHA. When no release is set, it falls back to the first 12 characters of the commit SHA. Invalid values are ignored rather than breaking your application.

createSeamwardService

createSeamwardService reports every Integration in a workspace with one ingest token. Issue the token under Settings, then Ingest token, in the mode you are setting up, and use each Integration's connection name from its Collector setup tab. See One token for every Integration.

import { createSeamwardService } from "@seamward/collector";
const seamward = createSeamwardService({  ingestToken: process.env.SEAMWARD_INGEST_TOKEN!,});
export const paystackWebhook = seamward.connection("paystack-webhook");export const crawlerApi = seamward.connection("crawler-api");

It takes ingestToken (sw_svc_v1...) and every createSeamwardCollector option above except connectionKey. The options apply to each connection.

Each connection has its own bounded queue, so a slow or failing connection cannot hold back or drop another's observations.

A connection name is 2 to 64 lowercase letters, digits and single hyphens. Construction throws when the ingest token is malformed, and connection(name) throws when the name is; neither echoes the value. The token decides the mode: a Test token records into each Integration's Sandbox connection and a Live token into its Live connection. A name no Integration in the workspace has is rejected by the API with 401, which the collector counts as rejected in stats() and does not throw into your application.

Observe operations

The Connection key binds this collector to one Integration, so observe entry points need only operation metadata. An Integration has one direction, one protocol, and its own ingest token. A process that observes multiple boundaries creates one collector for each Integration using that Integration's Connection key and ingest token.

observeWebhook

Wraps an inbound handler. Your handler's return value or thrown error stays authoritative; observation happens after the fact.

const handleCandidateWebhook = seamward.observeWebhook(  {    routeTemplate: "/webhooks/candidates",  },  async (payload) => {    const candidate = await candidateService.accept(payload);    return {      statusCode: 202,      eventType: "candidate.create",      outcome: {        accepted: true,        businessObjectType: "candidate",        businessObjectId: candidate.id,      },      correlation: { sourceEventId: candidate.providerEventId },    };  },);

Meta fields include routeTemplate (a stable template, never a concrete id), plus optional method (default POST), attempt, and correlation. Webhook observations use payloadLocation: "message". Set attempt to the provider delivery attempt and pass an idempotency key in correlation when one exists; both feed retry analysis without exposing the raw identifier. The handler may return a WebhookResult with optional statusCode, eventType, outcome, and correlation. Identifier fields (sourceEventId, idempotencyKey, businessObjectId) are converted to keyed hashes inside your process; Seamward never receives the original values. A thrown handler error is recorded as statusCode: 500, accepted: false, and then rethrown unchanged.

observeFetch

Wraps an outbound call with the standard fetch signature:

const providerFetch = seamward.observeFetch({  routeTemplate: "/v1/candidates",  eventType: "candidate.create",});
const response = await providerFetch(  "https://provider.example/v1/candidates",  init,);

The original response is returned as soon as the provider resolves; the collector clones it and inspects the JSON asynchronously, so response timing is preserved and non-JSON bodies still produce transport evidence. durationMs measures the provider call, not the inspection. Request headers are never read. A rejected fetch is recorded as statusCode: 0, accepted: false, and rethrown.

Fetch metadata also accepts optional method, attempt, correlation, payloadLocation, and requestPayload. Outbound fetch observations default to payloadLocation: "response".

Choose payloadLocation: "request" to describe the request contract instead. The collector inspects a JSON string in init.body or clones and parses a Request without consuming the original. For non-JSON bodies, provide a JSON representation through requestPayload; the collector derives its structural shape locally and never adds the values to the envelope. Without a parseable body or requestPayload, the operation still records transport evidence with a null payload shape.

observeCall

Observes one outgoing call made through any client, such as a provider SDK:

const intent = await seamward.observeCall(  operation,  () => stripe.paymentIntents.retrieve(paymentIntentId),  { reference: paymentIntentId },);

It resolves with the callback's own result and rethrows its own error. Synchronous and asynchronous callbacks are both accepted. Exactly one observation is recorded per call, as an outbound http-api operation.

Operation (ConnectedCallOperation<R>; define once and reuse):

Evidence (CallEvidence; what an extractor returns):

Invocation (CallInvocation; per call):

The built-in status reading checks status, statusCode, response.status, response.statusCode, lastResponse.statusCode, and $metadata.httpStatusCode, in that order, on results and errors. A thrown error whose code (or whose cause.code) is ECONNREFUSED, ECONNRESET, ENOTFOUND, ETIMEDOUT, EAI_AGAIN, EHOSTUNREACH, ENETUNREACH, EPIPE, UND_ERR_CONNECT_TIMEOUT, or UND_ERR_SOCKET is recorded as no response.

When no status is established the observation is recorded with status 0 and status state unknown; see Status state. An extractor that throws never affects the host call and increments extractionErrors. The Node collector guide has the full walkthrough.

observeQueuePublish

Wraps a broker-neutral outbound publish operation:

const publishCandidate = queuePublisherCollector.observeQueuePublish(  {    queueName: "candidate-events",    eventType: "candidate.requested",    attempt: 1,  },  async (message) => {    await queueClient.publish("candidate-events", message);    return { disposition: "acknowledged" as const };  },);

QueueMeta requires queueName, a stable queue, topic, or subscription label. It also accepts optional eventType, attempt, and correlation. A normal return defaults to disposition: "acknowledged" and accepted: true; an explicit disposition: "rejected" defaults to accepted: false. QueueResult.outcome can override that business result, and eventType or correlation returned by the publisher takes precedence over the metadata.

observeQueueConsumer

Wraps a broker-neutral inbound consumer with the same metadata and result types:

const consumeCandidate = queueConsumerCollector.observeQueueConsumer(  {    queueName: "candidate-events",    eventType: "candidate.received",  },  async (message) => {    await candidateService.accept(message);    return {      disposition: "acknowledged" as const,      outcome: { accepted: true },    };  },);

The publish and consumer wrappers do not acknowledge, retry, or dead-letter a broker message for you. Seamward does not currently ship broker-specific adapters. Your existing Kafka, RabbitMQ, SQS, Pub/Sub, or other client remains authoritative.

observeScheduledFeed

Wraps one scheduled import or export run:

const exportCandidates = feedExportCollector.observeScheduledFeed(  {    feedName: "nightly-candidate-export",    direction: "outbound",    eventType: "candidate.exported",  },  async (batch) => {    await feedClient.upload(batch);    return { disposition: "completed" as const };  },);

ScheduledFeedMeta requires a stable feedName and direction (inbound for imports, outbound for exports). It also accepts optional eventType, attempt, and correlation. A normal return defaults to disposition: "completed" and accepted: true; disposition: "rejected" defaults to accepted: false. ScheduledFeedResult can supply an explicit outcome, eventType, or correlation.

Feed metadata also accepts expectedIntervalMs: how often the feed is expected to run, as a whole number of milliseconds from 1 to 31,622,400,000 (366 days). It lets an overdue run be reported without history. Every wrapper's correlation accepts sequence, a provider-assigned whole number from 0 to 9,007,199,254,740,991 used to report deliveries that never arrived. Either value is left out of the envelope when unusable; the observation is still recorded.

Queue and feed compatibility projection

The queue and scheduled-feed entry points are protocol-native at the SDK boundary. Envelope 0.2 currently projects them into the shared transport object so existing APIs and workers remain compatible:

These status codes describe the collector's compatibility projection, not a broker or scheduler response. Use outcome for the business result and attempt for delivery or run retries.

Outcome amount and state

Wherever you return or pass an outcome, you may also report two facts about your own record. Reconciliation rules read them: amount-match compares amounts, and allowed-sequence checks the order of states.

return {  statusCode: 202,  eventType: "payout.paid",  outcome: {    accepted: true,    businessObjectType: "payout",    businessObjectId: payout.id,    amount: { value: payout.amountMinor, unit: "GBP_minor" },    state: payout.status,  },  correlation: { sourceEventId: event.id },};

These are values you choose to send, unlike everything else in an observation: they are stored as you report them. Report a total and a state name. Do not put an identifier, a name, or any other personal value in either field. An amount or state that does not meet the format is left out and the observation is still sent. An observation that reports either is sent as envelope 0.6. observeCall does not report them; use record or a handler result.

Account reference

If your records belong to your own customers, name the account in the outcome and an incident can tell you how many of your accounts it affects:

return {  statusCode: 202,  outcome: {    accepted: true,    businessObjectType: "application",    businessObjectId: application.id,    accountRef: application.employerId,  },};

accountRef is your own identifier for the customer or account, up to 512 characters. It is hashed in your process with your hash key, exactly like businessObjectId, and only the hash is sent as outcome.accountRefHash. Seamward never learns the identifier or a name; seamward hashes resolve turns the hashes back into your identifiers on your machine. With observeCall, pass accountRef in the invocation beside reference. An observation that names an account is sent as envelope 0.9.

Deprecation notices

Providers often announce a change in their response headers before it lands. The collector reads three kinds of header from an outbound call's response and sends what they say as dates and one version label, which Seamward turns into an advisory:

observeFetch reads the response's headers itself. observeCall reads them from where common clients keep them: headers on the result, on its response, or on its lastResponse, and the same places on a thrown error. For a client that keeps them elsewhere, return them from your extractor:

const retrieveCharge = {  integrationId: "int_payments",  method: "GET",  routeTemplate: "/v1/charges/:chargeId",  extract: (result: ProviderResult) => ({    statusCode: result.httpStatus,    payload: result.body,    headers: result.responseHeaders,  }),} satisfies CallOperation<ProviderResult>;

With record, pass them as responseHeaders.

No header is sent. Only the headers above are looked at, and a value is kept only when it parses to a date between 2000 and 2100 or to a version label of up to 64 letters, digits, ., _ or -. Anything else, including every other header, is discarded in your process. An observation that carries a notice is sent as envelope 0.7; see the envelope reference for the exact rules.

Rate limits and signature results

Rate limits. The collector reads a provider's rate-limit headers from the same places it reads deprecation notices and sends three whole numbers: the limit, what remains, and the seconds until it resets. Seamward uses them to warn when headroom stays low or the limit itself changes. Nothing needs configuring with observeFetch; with observeCall, a client that keeps headers somewhere unusual returns them from its extractor as headers.

The headers read are RateLimit-Limit, RateLimit-Remaining and RateLimit-Reset, the same names prefixed X-RateLimit- and X-Rate-Limit-, the combined RateLimit header, and Retry-After in seconds. A value that is not a whole number in range is discarded. No header is sent.

Signature results. If your webhook handler verifies the provider's signature, return the result and Seamward can tell a rotated signing secret from a general error spike:

const handleWebhook = collector.observeWebhook(  { integrationId: "int_payments", routeTemplate: "/webhooks/payments" },  async (payload) => {    const signatureValid = verifyProviderSignature(rawBody, signatureHeader);    if (!signatureValid) return { statusCode: 401, signatureValid };    await applyEvent(payload);    return { statusCode: 202, signatureValid };  },);

signatureValid is a boolean. Never pass the signature, the secret, or the header: there is no field for them. With record, set outcome.signatureValid. An observation that carries a rate limit or a signature result is sent as envelope 0.8.

record

Use record(input) for an operation that cannot be wrapped. It accepts direction, protocol, method, routeTemplate, statusCode, and durationMs, plus optional payloadLocation, attempt, eventType, payload, correlation, and outcome. The default payload location is message for webhooks, queues, and scheduled feeds; outbound HTTP API calls default to response, and inbound HTTP API calls default to request.

statusCode must be 0 or an integer from 100 through 599. Zero means that no protocol response was received. When outcome is omitted, accepted defaults to false for status 0 and to statusCode < 400 otherwise.

Lifecycle and stats

process.once("SIGTERM", async () => {  await seamward.stop();  process.exit(0);});

• flush() waits for the current bounded inspections, then starts or joins one bounded shipping cycle. Never rejects.
• stop() clears the flush timer and performs one final flush. Never rejects.
• stats() returns these counters: enqueued, shipped, rejected, dropped, failedBatches, queueLength (live), buildErrors, skippedInspections, and pendingInspections (live). The release that adds observeCall also adds extractionErrors, the number of call extractors that threw, and profileErrors, the number of value profiles that could not be built.

The flush timer is unreferenced, so the collector never keeps your process alive on its own.

Failure behavior

The design rule is fail-open: telemetry must never break the host application.

Raw payload value size does not determine envelope size because values never ship. Deep or unusually wide structures can still produce large envelopes, so normal ingestion request limits apply. A persistently failing endpoint retries on every flush until the queue overflows into drop-oldest, so watch failedBatches in your health checks.

Advanced exports

Most applications never need these; they exist for self-hosted setups and custom pipelines.

• createCollector(config): the unauthenticated core with the same surface, for callers managing raw scope ids and a complete policy themselves.
• createShipper(config): batching, signing, and retry transport alone, for shipping pre-built envelopes.
• buildEnvelope(input, policy, deps?): the pure envelope builder; deterministic with injected clock and id.
• redactPayload(value, policy) / redactHeaders(headers, policy) / keyedHash(value, key): standalone redaction helpers for your own logging pipelines.
• resolveDeploymentContext(explicit?, env?): the deployment detection used at construction.

Upload queue accounting

maxQueueSize bounds pending envelopes (default 1,000). One additional batch of up to maxBatchSize envelopes (default 50) can be in flight. Overflow drops the oldest pending envelopes and increments dropped; it cannot alter an active batch. A failed batch is returned to the front of the pending queue, where the same overflow rule applies. queueLength includes pending and in-flight envelopes. Acknowledgements update only the batch actually sent.

Node collector resource limits

These additive options apply to the Node collector:

Values must be positive integers within the maximum; invalid values use the default. Inspection also limits nesting to 32 levels and visits to 8,192 JSON nodes. Only JSON content types (application/json and +json) are read from HTTP bodies. Oversized, unsupported and timed-out bodies produce transport-only evidence where capacity allows. A full inspection queue skips the observation. skippedInspections counts these skips and pendingInspections reports active work. No raw payload is transmitted.

Envelope 0.3 carries payload.schemaInspected: false for a skipped body. Its null placeholder is excluded from structural drift detection; status and timing still contribute to behavioral analysis. Older observations without the marker retain their existing interpretation. Deploy the updated API and worker, then update external strict readers, before upgrading Node collectors to envelope 0.3. Older parsers reject the new wire version. Existing 0.1 and 0.2 data remains readable. See the wire contract for which versions each collector emits.

The collector emits envelope 0.4 or later. Version 0.4 adds a readable display route template for routes written with named parameters (:id or {id}) and an explicit status state for calls whose status could not be read. Operation identity is unchanged, so existing baselines continue.

Concurrent flush callers share a shipping cycle. Each cycle processes at most the queue size it started with; new observations wait for the next cycle. stop() prevents new inspection work, waits for the bounded current inspections, and attempts a final shipping cycle. With defaults, it finishes within roughly 11 seconds plus event-loop scheduling time. Timed-out uploads remain queued for a future flush; after process exit, unshipped in-memory observations are lost.

Next steps

• Node collector setup: install and record a first observation.
• Configuration: environment variables and option reference.
• Ingest observations: the wire contract the collector implements.