Event Subscribers

autotel-subscribers sends events to multiple product analytics platforms simultaneously.

Track once, send everywhere:

Primary metrics → OpenTelemetry
Product events → PostHog / Mixpanel / Amplitude
Customer data → Segment
Custom integrations → Webhooks

Installation

npm install autotel autotel-subscribers
npm install posthog-node    # For PostHog
npm install mixpanel        # For Mixpanel

Quick Start

import { init, track } from 'autotel';
import { PostHogSubscriber } from 'autotel-subscribers/posthog';

init({
  service: 'my-app',
  events: {
    subscribers: [new PostHogSubscriber({ apiKey: process.env.POSTHOG_KEY! })],
  },
});

track('user.signup', { userId: '123', plan: 'pro' });

Built-in Subscribers

Subscriber	Import
PostHog	`autotel-subscribers/posthog`
Mixpanel	`autotel-subscribers/mixpanel`
Segment	`autotel-subscribers/segment`
Amplitude	`autotel-subscribers/amplitude`
Webhook	`autotel-subscribers/webhook`
Slack	`autotel-subscribers/slack`
Security	`autotel-subscribers/security`
ArchitectureSnapshot	`autotel-subscribers/architecture-snapshot`

ArchitectureSnapshotSubscriber

Records every event your code emits — names, field paths, runtime types, sample values, producer/consumer edges — into a single snapshot. The output feeds autotel-eventcatalog to diff what production does against what your EventCatalog says it does.

import { init } from 'autotel';
import { ArchitectureSnapshotSubscriber } from 'autotel-subscribers/architecture-snapshot';

const snapshot = new ArchitectureSnapshotSubscriber({ service: 'my-app' });

init({ service: 'my-app', subscribers: [snapshot] });

// ...run integration tests, then persist the snapshot...
await writeFile('snapshot.json', JSON.stringify(snapshot.toSnapshot(), null, 2));

For each tracked event the snapshot includes fieldStats — observed runtime types and sampled primitive values per dotted field path — which powers type-drift and value-drift detection downstream.

Custom Subscriber

import { EventSubscriber, EventPayload } from 'autotel-subscribers';

class MySubscriber extends EventSubscriber {
  readonly name = 'MySubscriber';

  protected async sendToDestination(payload: EventPayload): Promise<void> {
    await fetch('https://api.example.com/events', {
      method: 'POST',
      body: JSON.stringify(payload),
    });
  }
}

Factories

autotel-subscribers/factories builds the built-in subscribers from a config object, so you skip the new and the import of each class. composeSubscribers combines several into one with a strategy (parallel, failover, round-robin, race, mirrored).

import {
  createPostHogSubscriber,
  createWebhookSubscriber,
} from 'autotel-subscribers/factories';

init({
  service: 'my-service',
  subscribers: [
    createPostHogSubscriber({ apiKey: 'phc_...' }),
    createWebhookSubscriber({ url: 'https://example.com/events' }),
  ],
});

Factories: createPostHogSubscriber, createMixpanelSubscriber, createAmplitudeSubscriber, createSegmentSubscriber, createWebhookSubscriber, createSlackSubscriber, createMockSubscriber, and composeSubscribers.

Streaming subscriber

For Kafka, Kinesis, or Pub/Sub, extend StreamingEventSubscriber instead of EventSubscriber. It adds batching, partitioning, buffer-overflow strategies (drop, block, disk), and backpressure.

import { StreamingEventSubscriber } from 'autotel-subscribers';

class KafkaSubscriber extends StreamingEventSubscriber {
  protected getPartitionKey(payload) {
    return payload.attributes?.userId ?? 'default';
  }

  protected async sendBatch(events) {
    await this.producer.send({
      topic: 'events',
      messages: events.map((e) => ({
        key: this.getPartitionKey(e),
        value: JSON.stringify(e),
      })),
    });
  }
}

Middleware

autotel-subscribers/middleware wraps a subscriber with cross-cutting behavior. applyMiddleware takes the subscriber and an array of middlewares, applied in order.

import {
  applyMiddleware,
  retryMiddleware,
  samplingMiddleware,
  enrichmentMiddleware,
} from 'autotel-subscribers/middleware';

const subscriber = applyMiddleware(new MySubscriber(), [
  retryMiddleware({ maxRetries: 3 }),
  samplingMiddleware(0.5), // keep 50% of events
  enrichmentMiddleware((event) => ({
    ...event,
    attributes: { ...event.attributes, region: 'us-east-1' },
  })),
]);

Available middleware:

Middleware	Purpose
`retryMiddleware({ maxRetries, delayMs })`	Retry failed sends with backoff.
`samplingMiddleware(rate)`	Keep a fraction of events (`0`–`1`).
`enrichmentMiddleware(fn)`	Add or rewrite event data before sending.
`filterMiddleware(predicate)`	Drop events that fail the predicate.
`transformMiddleware(fn)`	Replace the event shape.
`batchingMiddleware({ ... })`	Group events for bulk dispatch.
`rateLimitMiddleware({ requestsPerSecond })`	Cap throughput.
`circuitBreakerMiddleware({ ... })`	Stop sending past an error threshold.
`timeoutMiddleware({ timeoutMs })`	Reject sends that run too long.
`loggingMiddleware({ prefix })`	Log event flow for debugging.

Examples

example-subscribers — PostHog, Slack, and webhook subscribers with event filtering and funnel tracking.

Entry Points

autotel-subscribers — Base class + re-exports
autotel-subscribers/posthog — PostHog
autotel-subscribers/mixpanel — Mixpanel
autotel-subscribers/segment — Segment
autotel-subscribers/amplitude — Amplitude
autotel-subscribers/webhook — Webhook
autotel-subscribers/slack — Slack
autotel-subscribers/security — Forward security.* events to a webhook/SIEM/pager (Security Observability)
autotel-subscribers/architecture-snapshot — Architecture snapshot for drift detection
autotel-subscribers/factories — create*Subscriber builders + composeSubscribers
autotel-subscribers/middleware — Middleware composition
autotel-subscribers/testing — Test harnesses

StreamingEventSubscriber is re-exported from the package root (autotel-subscribers) alongside the base EventSubscriber.