Node collector

Set up @seamward/collector in a Node.js backend and record observations from one integration. The collector is observe-only: it batches redacted evidence asynchronously and never sits on your critical request path.

If you are not sure where to add the collector, start with assisted codebase setup. The setup agent discovers likely HTTP, webhook, queue, and scheduled-feed boundaries locally, creates a reviewable plan, previews every source change, and applies only the changes you approve.

Before you begin

• Node.js 22 or later; the package is ESM-only.
• A public Connection key and secret ingest token from the integration's Collector setup tab.

Install

npm install @seamward/collector

The collector is a prerelease (0.1.0-alpha); pin the version through your lockfile as with any dependency.

Configure

Set the connection values in your backend environment. This example observes five Integration boundaries in one process:

SEAMWARD_WEBHOOK_CONNECTION_KEY=sw_conn_v1.replace_webhookSEAMWARD_HTTP_API_CONNECTION_KEY=sw_conn_v1.replace_http_apiSEAMWARD_QUEUE_PUBLISH_CONNECTION_KEY=sw_conn_v1.replace_queue_publishSEAMWARD_QUEUE_CONSUME_CONNECTION_KEY=sw_conn_v1.replace_queue_consumeSEAMWARD_FEED_EXPORT_CONNECTION_KEY=sw_conn_v1.replace_feed_exportSEAMWARD_WEBHOOK_INGEST_TOKEN=sw_ing_replace_webhookSEAMWARD_HTTP_API_INGEST_TOKEN=sw_ing_replace_http_apiSEAMWARD_QUEUE_PUBLISH_INGEST_TOKEN=sw_ing_replace_queue_publishSEAMWARD_QUEUE_CONSUME_INGEST_TOKEN=sw_ing_replace_queue_consumeSEAMWARD_FEED_EXPORT_INGEST_TOKEN=sw_ing_replace_feed_export

Create one collector per Integration boundary and reuse that instance across operations belonging to the boundary:

import { createSeamwardCollector } from "@seamward/collector";
const connected = (connectionKey: string, ingestToken: string) =>  createSeamwardCollector({    connectionKey,    ingestToken,    deployment: {      service: "candidate-api",      release: process.env.RELEASE_VERSION,      commitSha: process.env.GITHUB_SHA,    },  });
export const webhookCollector = connected(  process.env.SEAMWARD_WEBHOOK_CONNECTION_KEY!,  process.env.SEAMWARD_WEBHOOK_INGEST_TOKEN!,);export const httpApiCollector = connected(  process.env.SEAMWARD_HTTP_API_CONNECTION_KEY!,  process.env.SEAMWARD_HTTP_API_INGEST_TOKEN!,);export const queuePublisherCollector = connected(  process.env.SEAMWARD_QUEUE_PUBLISH_CONNECTION_KEY!,  process.env.SEAMWARD_QUEUE_PUBLISH_INGEST_TOKEN!,);export const queueConsumerCollector = connected(  process.env.SEAMWARD_QUEUE_CONSUME_CONNECTION_KEY!,  process.env.SEAMWARD_QUEUE_CONSUME_INGEST_TOKEN!,);export const feedExportCollector = connected(  process.env.SEAMWARD_FEED_EXPORT_CONNECTION_KEY!,  process.env.SEAMWARD_FEED_EXPORT_INGEST_TOKEN!,);

deployment is optional; the collector also detects common platform variables (GITHUB_SHA, VERCEL_GIT_COMMIT_SHA, and others listed in configuration) automatically, and ignores invalid values instead of breaking your application.

Instrument a webhook

Wrap the existing business handler. Your handler's result or thrown error stays authoritative; observation happens after the operation.

const handleCandidateWebhook = webhookCollector.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 },    };  },);
fastify.post("/webhooks/candidates", async (request, reply) => {  const result = await handleCandidateWebhook(    request.body as Record<string, unknown>,  );  return reply.code(result.statusCode ?? 200).send();});

The identifier fields (sourceEventId, businessObjectId) are converted to keyed hashes inside your process before anything is queued; Seamward never receives the original values. For outbound calls, observeFetch wraps the standard fetch signature the same way.

Observe an outbound HTTP API

Wrap the provider call with observeFetch. Response-side evidence is the default, so the collector clones the response and inspects JSON asynchronously while returning the original Response as soon as the provider resolves.

const providerFetch = httpApiCollector.observeFetch({  routeTemplate: "/v1/candidates",  eventType: "candidate.create",});
const response = await providerFetch("https://provider.example/v1/candidates", {  method: "POST",  headers: { "content-type": "application/json" },  body: JSON.stringify({ externalId: "candidate_123" }),});

Use payloadLocation: "request" when the request contract is the useful side of the operation. JSON string bodies and cloned Request objects are inspected automatically. For form data, streams, protobuf, or another non-JSON body, provide a local JSON value through requestPayload:

const providerFetch = httpApiCollector.observeFetch({  routeTemplate: "/v1/documents",  eventType: "document.upload",  payloadLocation: "request",  requestPayload: {    documentId: "shape-only-local-value",    contentType: "application/pdf",  },});

requestPayload is used inside your process to calculate the structural shape. Its values are not added to the envelope. If no JSON representation is available, the operation still records transport evidence with a null payload shape. A network failure that receives no HTTP response records statusCode: 0, accepted: false, then rethrows the original fetch error.

Observe a provider SDK call

When a provider call goes through that provider's SDK, or any client other than fetch, wrap it with observeCall. Describe the operation once, then pass that description wherever the call is made.

import type { ConnectedCallOperation } from "@seamward/collector";
// Your code: one definition per provider operation.const retrievePaymentIntent: ConnectedCallOperation = {  method: "GET",  routeTemplate: "/v1/payment_intents/:paymentIntentId",  businessObjectType: "payment_intent",};
// Your code: the helper your services call instead of the SDK directly.export function retrieveObservedPaymentIntent(paymentIntentId: string) {  return httpApiCollector.observeCall(    retrievePaymentIntent,    () => stripe.paymentIntents.retrieve(paymentIntentId),    { reference: paymentIntentId },  );}

retrievePaymentIntent and retrieveObservedPaymentIntent are your own code, not collector exports. The only collector API here is observeCall. It resolves with the SDK's own result, rethrows the SDK's own error, and records exactly one observation either way, so callers cannot tell it is there.

The three arguments carry different things:

• The operation is stable. method and routeTemplate are the provider's HTTP method and route pattern for this SDK method, taken from the provider's API reference. Write parameters by name (:paymentIntentId). Never put a concrete id or a query string here.
• The callback is the call you already make, with its real arguments.
• The invocation carries what changes per call. reference is the id of the record involved; it leaves your process only as a keyed hash.

In the console the observation reads GET /v1/payment_intents/:paymentIntentId. Every call shares one operation whatever id it used.

How the status is read

An SDK result is not a Response, so the collector looks for the HTTP status in the places common clients put it, on results and on thrown errors alike:

Only that number is read. No other property of the result or error is touched or sent.

When the status is somewhere else, or you want the response shape watched for drift, add extractors to the operation:

import type { ConnectedCallOperation } from "@seamward/collector";
interface ProviderResult<T> {  http: { code: number };  body: T;}
const listInvoices: ConnectedCallOperation<ProviderResult<unknown>> = {  method: "GET",  routeTemplate: "/v2/customers/:customerId/invoices",  extract: (result) => ({    statusCode: result.http.code,    payload: result.body,  }),  extractError: (error) =>    error instanceof ProviderTimeout ? { noResponse: true } : {},};

payload is used inside your process to calculate the structural shape; its values never enter the envelope. Return noResponse: true only when you are certain no HTTP response arrived.

What is recorded when things go wrong

A status is never invented. An observation with an unknown status is still recorded, shows as Unknown in the console, and is left out of error-rate findings because it is not evidence of failure. If an extractor of yours throws, the host call is unaffected, the observation is recorded with an unknown status, and stats().extractionErrors increases.

Two limits to know:

• One observation per call. If the SDK retries internally, the observation still reports one attempt, and its duration covers every retry. Retry findings therefore do not apply to SDK-wrapped calls.
• Transport is not business success. Retrieving a payment successfully does not mean the payment succeeded. Return accepted from extract when the result tells you the business outcome.

Observe a queue

Use the publish wrapper for messages your service sends and the consumer wrapper for messages it receives. Both preserve the return value and rethrow host errors after recording them.

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

queueName must be a stable queue, topic, or subscription label, never a message id. Return disposition: "rejected" to record an explicit rejection. Set attempt from broker delivery metadata when it is available.

The wrappers are broker-neutral. Seamward does not currently ship adapters for Kafka, RabbitMQ, SQS, Pub/Sub, or another broker. Keep acknowledgement, visibility timeout, retry, and dead-letter behavior in your existing client.

Observe a scheduled feed

Wrap one import or export run and state its direction. An inbound feed brings provider data into your application; an outbound feed sends application data to the provider.

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

To have an overdue run reported, declare how often the feed runs:

const exportCandidates = feedExportCollector.observeScheduledFeed(  {    feedName: "hourly-candidate-export",    direction: "outbound",    expectedIntervalMs: 60 * 60 * 1000,  },  async (batch) => {    await feedClient.upload(batch);    return { disposition: "completed" as const };  },);

With a declared interval, Seamward reports a delivery gap once a run is more than one interval overdue, on Live connections, without first learning the schedule from history. The interval is a whole number of milliseconds up to 366 days.

Use a stable feedName, never a run id or timestamp. Return disposition: "rejected" for a completed run whose business result was rejected. Thrown job errors are recorded and rethrown unchanged, so your scheduler remains responsible for retry and failure handling.

Queue and scheduled-feed wrappers expose protocol-native options, but envelope 0.2 still stores them through a compatibility projection: the queue or feed label occupies transport.routeTemplate, method is POST, and payloadLocation is message. This keeps current APIs and workers compatible. Treat these transport fields as storage details, not broker commands.

Describe field values

By default the collector also describes each field's value, so Seamward can notice a change that keeps the shape and moves the meaning, such as an amount switching from minor to major units. It sends a format label and bands for size and length, never the value. Nothing in your instrumentation changes; set valueProfile only to turn it down or up:

const seamward = createSeamwardCollector({  connectionKey: process.env.SEAMWARD_CONNECTION_KEY!,  ingestToken: process.env.SEAMWARD_INGEST_TOKEN!,  valueProfile: "off", // or "structural" (the default), or "full"});

Fields named in your redaction policy are never described. full additionally sends keyed hashes of short values and requires a hashKey that only you hold. Redaction states exactly what each level sends, and Value drift lists what is detected.

Verify

1. Start the backend with the Connection key and ingest token configured.
2. Trigger the handler once with a safe test payload.
3. Wait for the five-second automatic flush, or call await webhookCollector.flush().
4. Open the integration in Seamward and confirm the setup status reports the first observation instead of No observation received yet.

If nothing arrives, seamward.stats() tells you where it stopped; the troubleshooting guide maps each counter to its fix.

Shutdown

Give queued observations a final flush during controlled shutdown:

process.once("SIGTERM", async () => {  await Promise.all([    webhookCollector.stop(),    httpApiCollector.stop(),    queuePublisherCollector.stop(),    queueConsumerCollector.stop(),    feedExportCollector.stop(),  ]);  process.exit(0);});

flush() and stop() never reject, and the flush timer never keeps your process alive on its own.

Bounded inspection and shutdown

The collector caps pending inspections at 64, reads at most 256,000 bytes per JSON body and stops a stalled inspection after one second by default. Check skippedInspections and pendingInspections alongside delivery counters. stop() finishes its current inspection and shipping cycle within roughly 11 seconds with defaults, plus event-loop scheduling time. Allow that grace period during shutdown. See the resource limits for configuration and the API-before-collector upgrade order.

Next steps

• Observe webhooks well: boundaries, route templates, and the outcome contract.
• Collector SDK reference: exact metadata, result types, and status projections for every wrapper.
• Observation envelope: how the four protocols appear on the wire.
• Redaction: exactly what leaves your process.