autotel-subscribers

Send events to multiple platforms

Subscribers for autotel to send events to PostHog, Mixpanel, Amplitude, Segment, and custom webhooks.

Why Use This?

Track once, send everywhere:
- Primary metrics → OpenTelemetry (infrastructure monitoring)
- Product events → PostHog / Mixpanel / Amplitude
- Customer data → Segment
- Custom integrations → Webhooks (Zapier, Make.com, etc.)

Zero overhead when not used:
Adapters are optional. If you don't use them, they're tree-shaken out (0 bytes).

---

Building Custom Subscribers

Two base classes available:

$3

Use for most custom subscribers. Provides production-ready features:

- Error handling (automatic catching + custom handlers)
- Pending request tracking (ensures delivery during shutdown)
- Graceful shutdown (drains pending requests)
- Enable/disable control (runtime toggle)

When to use: Any custom adapter (HTTP APIs, databases, webhooks, etc.)

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

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

protected async sendToDestination(payload: EventPayload): Promise {
// Send to your platform
await fetch('https://api.example.com/events', {
method: 'POST',
body: JSON.stringify(payload)
});
}
}
`

$3

Extends EventSubscriber with batching and partitioning for streaming platforms.

When to use: Kafka, Kinesis, Pub/Sub, event streams, high-volume data pipelines

`typescript
import { StreamingEventSubscriber } from 'autotel-subscribers';

class KafkaSubscriber extends StreamingEventSubscriber {
readonly name = 'KafkaSubscriber';

protected async sendBatch(events: EventPayload[]): Promise {
await this.producer.send({
topic: 'events',
messages: events.map(e => ({ value: JSON.stringify(e) }))
});
}
}
`

---

Installation

`bash


Core package (required)

pnpm add autotel

Subscribers package (optional)

pnpm add autotel-subscribers

Install the events SDKs you need

pnpm add posthog-node # For PostHog
pnpm add mixpanel # For Mixpanel
pnpm add @segment/events-node # For Segment
pnpm add @amplitude/events-node # For Amplitude
`

---

Quick Start

$3

Import subscribers directly from their entry points:

`typescript
import { Events } from 'autotel/events';
import { PostHogSubscriber } from 'autotel-subscribers/posthog';
import { WebhookSubscriber } from 'autotel-subscribers/webhook';

const events = new Event('checkout', {
subscribers: [
new PostHogSubscriber({ apiKey: process.env.POSTHOG_API_KEY! }),
new WebhookSubscriber({ url: 'https://your-webhook.com' })
]
});

// Sent to: OpenTelemetry + PostHog + Webhook
events.trackEvent('order.completed', { userId: '123', amount: 99.99 });
`

$3

Create an adapter in 25 lines:

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

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

constructor(private apiKey: string) {
super();
}

protected async sendToDestination(payload: EventPayload): Promise {
await fetch('https://your-api.com/events', {
method: 'POST',
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify(payload)
});
}
}

// Use it!
const events = new Event('my-app', {
subscribers: [new MySubscriber('your-api-key')]
});
`

That's it! Extend EventSubscriber and implement sendToDestination(). You get error handling, graceful shutdown, and pending request tracking automatically. See Your First Adapter Guide for details.

$3

`typescript
import { AdapterTestHarness } from 'autotel-subscribers/testing';

const harness = new AdapterTestHarness(new MySubscriber('test-key'));
const results = await harness.runAll();

AdapterTestHarness.printResults(results);
// All tests passed! Your adapter is ready to use.
`

$3

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

const subscriber = applyMiddleware(
new MySubscriber('api-key'),
[
retryMiddleware({ maxRetries: 3 }), // Retry failed requests
samplingMiddleware(0.1) // Only send 10% of events
]
);
`

---

Built-in Adapters

$3

`typescript
import { Events } from 'autotel/events';
import { PostHogSubscriber } from 'autotel-subscribers/posthog';

const events = new Event('checkout', {
subscribers: [
new PostHogSubscriber({
apiKey: process.env.POSTHOG_API_KEY!,
host: 'https://us.i.posthog.com' // optional
})
]
});

// Sent to: OpenTelemetry + PostHog
events.trackEvent('order.completed', {
userId: '123',
amount: 99.99
});
`

Serverless Configuration (AWS Lambda, Vercel, Next.js):

`typescript
const subscriber = new PostHogSubscriber({
apiKey: 'phc_...',
serverless: true, // Auto-configures for serverless (flushAt: 1, flushInterval: 0)
});
`

Browser Usage (with global PostHog client):

`typescript
// When PostHog is already loaded via script tag
const subscriber = new PostHogSubscriber({
useGlobalClient: true, // Uses window.posthog
});
`

Advanced Options:

`typescript
const subscriber = new PostHogSubscriber({
apiKey: 'phc_...',

// Automatic filtering (enabled by default)
filterUndefinedValues: true, // Removes undefined/null from attributes

// Enhanced error handling
onErrorWithContext: (ctx) => {
console.error(${ctx.eventType} failed: ${ctx.eventName}, ctx.error);
Sentry.captureException(ctx.error, { extra: ctx });
},
});
`

Custom Funnel Tracking:

`typescript
// Track custom step names (not limited to 'started'/'completed')
event.trackFunnelProgression('checkout', 'cart_viewed', 1);
event.trackFunnelProgression('checkout', 'shipping_selected', 2);
event.trackFunnelProgression('checkout', 'payment_entered', 3);
`

$3

`typescript
import { MixpanelSubscriber } from 'autotel-subscribers/mixpanel';

const events = new Event('checkout', {
subscribers: [
new MixpanelSubscriber({
token: process.env.MIXPANEL_TOKEN!
})
]
});
`

$3

`typescript
import { SegmentSubscriber } from 'autotel-subscribers/segment';

const events = new Event('checkout', {
subscribers: [
new SegmentSubscriber({
writeKey: process.env.SEGMENT_WRITE_KEY!
})
]
});
`

$3

`typescript
import { AmplitudeSubscriber } from 'autotel-subscribers/amplitude';

const events = new Event('checkout', {
subscribers: [
new AmplitudeSubscriber({
apiKey: process.env.AMPLITUDE_API_KEY!
})
]
});
`

$3

`typescript
import { WebhookSubscriber } from 'autotel-subscribers/webhook';

const events = new Event('checkout', {
subscribers: [
new WebhookSubscriber({
url: 'https://hooks.zapier.com/hooks/catch/...',
headers: { 'X-API-Key': 'secret' },
maxRetries: 3
})
]
});
`

---

Multi-Platform Tracking

Send to multiple platforms simultaneously:

`typescript
import { Events } from 'autotel/events';
import { PostHogSubscriber } from 'autotel-subscribers/posthog';
import { MixpanelSubscriber } from 'autotel-subscribers/mixpanel';
import { SegmentSubscriber } from 'autotel-subscribers/segment';

const events = new Event('checkout', {
subscribers: [
new PostHogSubscriber({ apiKey: 'phc_...' }),
new MixpanelSubscriber({ token: '...' }),
new SegmentSubscriber({ writeKey: '...' })
]
});

// Sent to: OpenTelemetry + PostHog + Mixpanel + Segment
events.trackEvent('order.completed', {
userId: '123',
amount: 99.99,
currency: 'USD'
});
`

---

Delivery Patterns

Autotel-subscribers provides direct subscribers (fire-and-forget) - events are sent immediately to events platforms.

$3

Simple, fire-and-forget tracking - Events sent immediately to events platforms:

`typescript
const events = new Event('app', {
subscribers: [new PostHogSubscriber({ apiKey: '...' })]
})

// Events sent immediately, real-time
events.trackEvent('user.signup', { userId: '123' })
events.trackEvent('page.viewed', { path: '/checkout' })
`

Use for:
- Page views, button clicks, feature usage
- User behavior tracking
- High-volume, non-critical events
- Real-time events dashboards

Benefits:
- Simple, zero infrastructure
- Real-time delivery
- No database overhead
- Fire-and-forget

Trade-offs:
- Events can be lost if adapter/network fails
- No atomicity with database transactions

$3

For guaranteed delivery with atomicity, use the separate autotel-outbox package.

This provides:
- Guaranteed delivery (retries on failure)
- Atomicity with database state changes
- Fan-out to multiple destinations
- Requires database table + publisher worker
- Adds latency (1+ minute delay)

Install:
`bash
npm install autotel-outbox
`

Usage:
`typescript
import { OutboxEventSubscriber } from 'autotel-outbox';
import { PostHogSubscriber } from 'autotel-subscribers/posthog';

const outbox = new DrizzleD1OutboxStorage(env.DB);
const events = new Event('checkout', {
subscribers: [
new OutboxEventSubscriber(outbox, { aggregateType: 'Order' })
]
});
`

---

Adapter Methods

All subscribers implement these methods:

`typescript
interface EventSubscriber {
// Track events
trackEvent(name: string, attributes?: Record): Promise;

// Track conversion funnels (enum-based steps)
trackFunnelStep(
funnelName: string,
step: 'started' | 'completed' | 'abandoned' | 'failed',
attributes?: Record
): Promise;

// Track funnel progression (custom step names)
trackFunnelProgression?(
funnelName: string,
stepName: string, // Any string, not limited to enum
stepNumber?: number, // Optional numeric position
attributes?: Record
): Promise;

// Track business outcomes
trackOutcome(
operationName: string,
outcome: 'success' | 'failure' | 'partial',
attributes?: Record
): Promise;

// Track business values (revenue, counts, etc.)
trackValue(
name: string,
value: number,
attributes?: Record
): Promise;

// Flush and clean up resources
shutdown?(): Promise;
}
`

---

Custom Subscriber

Create your own adapter for any platform:

`typescript
import { EventSubscriber } from 'autotel/events-adapter';

class MyCustomSubscriber implements EventSubscriber {
trackEvent(name: string, attributes?: Record): void {
// Send to your platform
fetch('https://api.myplatform.com/events', {
method: 'POST',
body: JSON.stringify({ event: name, ...attributes })
});
}

trackFunnelStep(funnel: string, step: string, attributes?: any): void {
// Implement funnel tracking
}

trackOutcome(operation: string, outcome: string, attributes?: any): void {
// Implement outcome tracking
}

trackValue(name: string, value: number, attributes?: any): void {
// Implement value tracking
}
}

// Use it
const events = new Event('app', {
subscribers: [new MyCustomSubscriber()]
});
`

---

Configuration

$3

`typescript
const events = new Event('checkout', {
subscribers: [
new PostHogSubscriber({
apiKey: 'phc_...',
enabled: process.env.NODE_ENV === 'production' // Only in prod
}),
new MixpanelSubscriber({
token: '...',
enabled: false // Temporarily disabled
})
]
});
`

$3

`typescript
const posthog = new PostHogSubscriber({ apiKey: 'phc_...' });
const segment = new SegmentSubscriber({ writeKey: '...' });

// Before app shutdown
await posthog.shutdown();
await segment.shutdown();
`

---

Tree-Shaking

Adapters are fully tree-shakeable:

`typescript
// Only PostHog code is bundled (not Mixpanel, Segment, etc.)
import { PostHogSubscriber } from 'autotel-subscribers/posthog';
`

Bundle sizes (gzipped):
- PostHog: ~8KB
- Mixpanel: ~6KB
- Segment: ~12KB
- Amplitude: ~10KB
- Webhook: ~2KB

---

Performance

Zero overhead when not used:
- If subscribers: [] (empty), no adapter code runs
- Tree-shaken out in production builds

Minimal overhead when used:
- Adapters only fire if added to the array
- Non-blocking (fire-and-forget)
- No impact on primary OpenTelemetry metrics

---

Middleware (Composition Patterns)

Add behaviors without modifying adapter code:

$3

`typescript
import {
applyMiddleware,
retryMiddleware, // Exponential backoff retry
samplingMiddleware, // Send only X% of events
enrichmentMiddleware, // Add fields to events
loggingMiddleware, // Debug events
filterMiddleware, // Only send matching events
transformMiddleware, // Transform events
batchingMiddleware, // Batch for efficiency
rateLimitMiddleware, // Throttle requests
circuitBreakerMiddleware, // Prevent cascading failures
timeoutMiddleware // Add timeouts
} from 'autotel-subscribers/middleware';
`

$3

Retry with Circuit Breaker:
`typescript
const subscriber = applyMiddleware(
new PostHogSubscriber({ apiKey: '...' }),
[
retryMiddleware({ maxRetries: 3, delayMs: 1000 }),
circuitBreakerMiddleware({ failureThreshold: 5, timeout: 60000 })
]
);
`

Sample Events (Reduce Costs):
`typescript
// Only send 10% of events
const subscriber = applyMiddleware(
new WebhookSubscriber({ url: '...' }),
[samplingMiddleware(0.1)]
);
`

Enrich Events:
`typescript
const subscriber = applyMiddleware(
adapter,
[
enrichmentMiddleware((event) => ({
...event,
attributes: {
...event.attributes,
environment: process.env.NODE_ENV,
timestamp: Date.now()
}
}))
]
);
`

Batch Events:
`typescript
const subscriber = applyMiddleware(
adapter,
[batchingMiddleware({ batchSize: 100, flushInterval: 5000 })]
);
`

---

Testing Custom Subscribers

$3

Validate your adapter works correctly:

`typescript
import { AdapterTestHarness } from 'autotel-subscribers/testing';

const harness = new AdapterTestHarness(new MySubscriber());
const results = await harness.runAll();

if (results.passed) {
console.log('All tests passed!');
} else {
console.error('Tests failed:', results.failures);
}

// Or use the built-in printer
AdapterTestHarness.printResults(results);
`

Tests include:
- Basic event tracking
- Funnel tracking
- Outcome tracking
- Value tracking
- Concurrent requests (50 events)
- Error handling
- Graceful shutdown

$3

Test webhook subscribers without real HTTP calls:

`typescript
import { MockWebhookServer } from 'autotel-subscribers/testing';

const server = new MockWebhookServer();
const url = await server.start();

const subscriber = new WebhookSubscriber({ url });
await subscriber.trackEvent('test', { foo: 'bar' });

// Assert
const requests = server.getRequests();
expect(requests).toHaveLength(1);
expect(requests[0].body.event).toBe('test');

await server.stop();
`

---

Package Exports

All exports available:

`typescript
// Import subscribers from their specific entry points
import { PostHogSubscriber } from 'autotel-subscribers/posthog';
import { MixpanelSubscriber } from 'autotel-subscribers/mixpanel';
import { SegmentSubscriber } from 'autotel-subscribers/segment';
import { AmplitudeSubscriber } from 'autotel-subscribers/amplitude';
import { WebhookSubscriber } from 'autotel-subscribers/webhook';
import { SlackSubscriber } from 'autotel-subscribers/slack';

// Base classes for building custom subscribers
import { EventSubscriber, EventPayload } from 'autotel-subscribers';
import { StreamingEventSubscriber } from 'autotel-subscribers';

// Middleware (composition)
import {
applyMiddleware,
retryMiddleware,
samplingMiddleware,
/ ... 8 more middleware functions /
} from 'autotel-subscribers/middleware';

// Testing utilities
import {
AdapterTestHarness,
MockWebhookServer,
MockEventSubscriber
} from 'autotel-subscribers/testing';

// For outbox pattern, see autotel-outbox package
``

---

Resources

- Your First Adapter Guide - Create a custom adapter in 5 minutes
- Quickstart Template - Copy-paste 20-line template
- Testing Guide - Validate your adapter works
- Middleware Guide - Add retry, sampling, etc.
- Outbox Pattern - For transactional outbox pattern

---

Examples

See the autotel-examples package for complete examples.

---

License

MIT © Jag Reehal