@metrics/node

• Metrics client. It has the same surface as Segment's analytics-node (track, identify, page, screen, group, alias, flush, close). It validates messages locally, sends them in batches, signs requests and retries failures on the Tracking API schedule.
• Durable PostgreSQL outbox (@metrics/node/outbox-pg). It writes events inside your business transaction, and a dispatcher delivers them, so committed events survive restarts.
• metrics CLI. metrics pull generates TypeScript types and a typed track helper from your event catalogue. metrics validate checks sample events against the catalogue in CI.

Wire contract: docs/api/tracking-api.md (Segment Tracking API Spec compatible).

Install

pnpm add @metrics/node
pnpm add pg            # only if you use the PostgreSQL outbox

Node.js ≥ 24 (global fetch). Runtime dependency: @metrics/contracts (shared message schemas). pg is an optional peer dependency.

Quick start

import { Metrics, deterministicMessageId } from '@metrics/node';

const metrics = new Metrics({
  writeKey: process.env.METRICS_WRITE_KEY!,           // pk_live_… / pk_test_…
  signingSecret: process.env.METRICS_SIGNING_SECRET,  // sk_live_… (server only, optional)
  host: 'https://collect.example.com',
});

metrics.identify({ userId: 'usr_7c1e', traits: { plan: 'pro' } });
metrics.track({
  userId: 'usr_7c1e',
  event: 'purchase_completed',
  messageId: deterministicMessageId('purchase_completed', order.id),
  properties: { transaction_id: order.number, plan_id: 'pro', revenue: 49.9, currency: 'EUR' },
});

// On shutdown: send what is queued, stop the timers.
process.on('SIGTERM', async () => {
  await metrics.close();
  process.exit(0);
});

Calls never throw into your code. Invalid messages go to onRejected, and delivery problems go to onError. Every call also accepts a Segment-style callback, (err, messageId) => …, which is called with the message's final outcome.

Options

Every message gets these defaults: messageId (a random UUID if missing), timestamp (now, and Date values become ISO 8601), and context.library = { name: '@metrics/node', version }. sentAt is set on each batch request. Messages are validated with the shared contract before they are queued. A message over 32 KB is rejected locally with message_too_large.

ping() checks that the host is reachable and the write key is known (GET /v1/config/{writeKey}). The Tracking API has no authenticated no-op write, so a wrong signing secret only shows up as a 401 on the first send.

Joining a browser visit

A server event (a signup, a payment webhook) can carry what the browser knew about the visit, so it gets the same session, campaign and click ids. The web SDK's metrics.getContext() returns it; send it to your server with the request (a header or a form field) and pass it on as the message context:

// Browser: fetch('/api/signup', { method: 'POST', body: JSON.stringify({ ...form, metricsContext: metrics.getContext() }) })
const visit = req.body.metricsContext; // { anonymousId, session, consent, attribution, clickIds, ga }
metrics.track({
  userId: user.id,
  anonymousId: visit?.anonymousId,
  event: 'signup_completed',
  messageId: deterministicMessageId('signup_completed', user.id),
  context: visit && {
    session: visit.session,           // the visit's session id
    consent: visit.consent,
    attribution: visit.attribution,   // first / last non-direct / session touch and `method`
    clickIds: visit.clickIds,         // only present with marketing consent
    ga: visit.ga,                     // Google Analytics client and session ids, only with analytics consent
  },
});

• attribution.method is consent when the browser kept the touches across visits and same_visit when it only saw this visit (no consent). The lead's attribution shows it. Another value is dropped, never rejected.
• ga ({ clientId, sessionId }, digits as GA writes them) lets the GA4 Measurement Protocol destination send the conversion into the visitor's GA4 session. An invalid value is dropped, never rejected.
• Forward only what the browser sent: the server never reads cookies for you, and no consent is implied.

Request signing

When signingSecret is set, each request carries this header:

Metrics-Signature: t=<unix seconds>,v1=hex(HMAC-SHA256(signingSecret, `${t}.${rawBody}`))

The signature is computed over the exact bytes sent. The server accepts ±300 s of clock skew, so keep the server clock synced (NTP). A 401 usually means a wrong key or secret, or a clock that is off. Keep sk_… secrets on the server only.

Idempotency

The Tracking API stores each messageId + timestamp pair once. Give every business event an id derived from the real-world occurrence, so retries, duplicate webhooks and job re-runs are harmless:

deterministicMessageId('purchase_completed', order.id);                        // one purchase per order
deterministicMessageId('subscription_renewed', sub.id, period.start.toISOString());
deterministicMessageId('trial_started', account.id);                           // once per account

The result is ${event}:${sha256(parts.join('|')).slice(0, 32)}, so internal ids are hashed and never sent in clear text. Pass the same timestamp for the same occurrence (the business time, not "now") when an event can be emitted twice. Never use e-mail addresses or phone numbers as userId; use opaque ids.

Delivery and retries

Batches stay within the Tracking API limits: at most 500 messages and 500 KB per request.

In-memory retries are best-effort. They live in the process and are lost when it exits or crashes. close() makes a final attempt for everything in memory and drops what still fails. For events that must not be lost (signups, purchases, renewals…), use the outbox.

Durable outbox (PostgreSQL)

The outbox writes the event in the same transaction as the business change, so the event is committed or rolled back together with it. A dispatcher then delivers it with the semantics above.

import pg from 'pg';
import { deterministicMessageId } from '@metrics/node';
import { enqueue, OutboxDispatcher, outboxDDL } from '@metrics/node/outbox-pg';

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

// Once (or in a migration): idempotent CREATE TABLE / INDEX IF NOT EXISTS.
// Table "metrics_outbox". For another name use outboxDDL('app.metrics_outbox') and pass the same `table` option
// to enqueue() and OutboxDispatcher.
await pool.query(outboxDDL());

// Inside the business transaction:
const client = await pool.connect();
try {
  await client.query('BEGIN');
  const order = await createOrder(client, input);
  await enqueue(client, {
    type: 'track',
    userId: order.userId,
    event: 'purchase_completed',
    messageId: deterministicMessageId('purchase_completed', order.id),
    timestamp: order.paidAt,
    properties: { transaction_id: order.number, plan_id: order.planId, revenue: order.total, currency: order.currency },
  });
  await client.query('COMMIT');
} catch (err) {
  await client.query('ROLLBACK');
  throw err;
} finally {
  client.release();
}

// In every API / worker process:
const dispatcher = new OutboxDispatcher({
  pool,
  writeKey: process.env.METRICS_WRITE_KEY!,
  signingSecret: process.env.METRICS_SIGNING_SECRET,
  host: 'https://collect.example.com',
});
dispatcher.start();                    // a tick now, then every 30 s
process.on('SIGTERM', () => void dispatcher.stop());

• enqueue validates the message (with the same rules as track) and inserts it with ON CONFLICT (message_id) DO NOTHING. It returns { ok, messageId, inserted }, where inserted is false for a duplicate. It never throws and never aborts your transaction: the insert runs inside a SAVEPOINT, so a missing table or a permission error is rolled back to the savepoint and logged. Pass { strict: true } to get exceptions instead. With a pg.Pool (autocommit) the savepoint is skipped. On a plain client outside a transaction, PostgreSQL refuses the savepoint and the insert runs on its own; pass { savepoint: false } there to skip the extra round trip.
• The dispatcher leases due rows with UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP LOCKED) SET next_attempt_at = now() + lease, so any number of processes can run it. It holds no database connection during HTTP. If a worker crashes, its lease expires (default 300 s) and another worker resends the rows; the Tracking API deduplicates by messageId. While leases come back full, each tick drains up to maxBatchesPerTick (10) leases of batchSize (100) rows. drain() runs a tick on demand, for example right after a commit or from a cron job instead of start().
• Row status:
  • pending: waiting, or retrying (attempts, next_attempt_at, last_error).
  • sent: accepted (sent_at).
  • rejected: the Tracking API refused the message. Fix the producer; do not resend unchanged.
  • abandoned: 403, or the 24 h retry window was exhausted.

  Abandoned rows can be re-queued after the cause is fixed: UPDATE metrics_outbox SET status = 'pending', next_attempt_at = now(), created_at = now() WHERE status = 'abandoned'. The 31-day timestamp limit of the Tracking API still applies.

• Housekeeping is up to you, for example a daily DELETE FROM metrics_outbox WHERE status IN ('sent', 'rejected') AND created_at < now() - interval '30 days'.

CLI

metrics pull: typed events from the catalogue

export METRICS_API_TOKEN=mk_…                         # API key with management:read
npx metrics pull --app app_123 --host https://api.example.com \
  --out src/metrics.generated.ts --bundle-out metrics.schema.json
npx metrics pull --from-file metrics.schema.json      # offline

The command calls GET {host}/v1/apps/{appId}/schema (Authorization: Bearer <token>), validates the bundle and writes:

• an interface per event, with required and optional properties and JSDoc from the descriptions, roles and constraints;
• MetricsEvents (event name → properties), MetricsIdentifyTraits and MetricsGroupTraits;
• createTypedMetrics(client), a typed facade over the client.

import { createTypedMetrics } from './metrics.generated.js';

const typed = createTypedMetrics(metrics);
typed.track('signup_completed', { signup_method: 'email' }, { userId: 'usr_7c1e' });
// typed.track('signup_completed', { signup_method: 'fax' }, …)  ← compile error

Types map as follows:

• string, datetime (ISO 8601) and currency (ISO 4217) become string.
• number, integer and money become number.
• boolean stays boolean.
• enum becomes a union of literals.
• object becomes Record<string, unknown>.
• array becomes T[].

Blocked events are left out, and deprecated ones are marked @deprecated. The output is deterministic, so commit it and re-pull when the catalogue changes.

metrics validate: check sample events in CI

npx metrics validate --schema metrics.schema.json test/fixtures/events.jsonl

Each line of the JSON Lines file is one message. A .json file can hold an array, a { "batch": [...] } body or a single message. Each message is checked for:

• the Spec shape;
• size (≤ 32 KB);
• unknown events (an error in strict mode, or with --strict; a warning otherwise);
• blocked, deprecated and draft events;
• missing required properties, wrong types, enum values, maxLength and minimum/maximum;
• unplanned properties;
• identify and group traits.

The special properties of the Tracking API (revenue, currency, tax, net_revenue, order_id, transaction_id, plan_id, value, test) are type-checked but never reported as unplanned. --format json prints a machine-readable report. The command exits with 1 on any error and 2 on usage or input errors.

Using a Segment client instead

The Tracking API also accepts @segment/analytics-node pointed at your collect host. That client cannot sign requests, does not report rejected[] and has no outbox. Use this SDK for sources that require signatures.