an opinionated event-driven framework for building asynchronous systems on amazon web services.
eventing-framework connects services through a typed event bus built on sns topics, sqs queues, and s3 storage. define domain events, run handlers when they arrive, and emit new events from anywhere in your code. the framework handles publishing, polling, oversized payloads, and s3 reference resolution behind the scenes.
- typed domain events with per-service base classes
- transparent oversized payload routing (inline or s3 reference)
- error isolation - a failing handler doesn't block the rest of the batch
eventing-framework lets your services communicate without coupling. publish what happened. subscribe to what matters.
npm install eventing-frameworkrequires node.js 20 or later.
extend DomainEvent to define your events. use an intermediate base class to inject per-service metadata once.
import { DomainEvent } from 'eventing-framework';
class OrderEvent<PayloadType = Record<string, unknown>> extends DomainEvent {
constructor(name: string, payload: PayloadType) {
super(name, { ...payload, app: 'order-service', category: 'order' });
}
}
class OrderPlaced extends OrderEvent<{ id: string; total: number }> {
constructor(payload: { id: string; total: number }) {
super('OrderPlaced', payload);
}
}
class OrderProcessed extends OrderEvent<{ id: string }> {
constructor(payload: { id: string }) {
super('OrderProcessed', payload);
}
}configure only what your service needs, topic and storage for the publishing side, queue and storage for the consuming side.
import { Application } from 'eventing-framework';
const app = new Application({
name: 'order-service',
environment: 'development',
topic: {
arn: 'arn:aws:sns:us-east-1:123456789012:order-events',
config: { region: 'us-east-1' },
},
queue: {
url: 'https://sqs.us-east-1.amazonaws.com/123456789012/notification-queue',
config: { region: 'us-east-1' },
params: { MessageAttributeNames: ['All'], WaitTimeSeconds: 5 },
},
storage: {
bucket: 'eventing-reference-bucket',
config: { region: 'us-east-1' },
},
scheduler: { interval: 5_000 },
});services consume events published by other services. the framework polls sqs, resolves the payload, and routes it to your handler.
use a typed event map for auto-typed payloads.
type AppEvents = {
OrderPlaced: { id: string; total: number };
OrderProcessed: { id: string };
};
const app = new Application<AppEvents>({ ... });
app.on('OrderPlaced', async (payload) => {
console.log(`order ${payload.id} for $${payload.total}`);
});or use an explicit generic on a bare Application().
const app = new Application({ ... });
app.on<{ id: string; total: number }>('OrderPlaced', async (payload) => {
console.log(`order ${payload.id} for $${payload.total}`);
});register the event, then emit from anywhere in your process.
import { EventPublisher } from 'eventing-framework';
app.register('OrderProcessed');
EventPublisher.emit(new OrderProcessed({ id: '123' }));a handler can emit the next event in the flow, the chain continues to downstream services without coordination.
app.on('OrderPlaced', async (payload) => {
await processOrder(payload);
EventPublisher.emit(new OrderProcessed({ id: payload.id }));
});await app.start(); // begins polling on the configured interval
await app.stop(); // stops the scheduler, clears handlers and registrationstwo working examples against localstack, no aws account needed. see examples for more details.
- single event - a publisher and a consumer communicating through a single event type
- multi-step workflow - four services chained across three event types, with oversized payloads and error handling
sns handles fan-out to multiple queues. sqs provides at-least-once delivery with visibility timeouts for retry. s3 stores payloads that exceed sns's 256kb message size limit, the framework routes oversized payloads transparently, consumers resolve references without knowing the difference.
sns-to-sqs subscriptions must use RawMessageDelivery: true.
the framework does not delete s3 objects after consumption. configure an s3 lifecycle policy on your reference bucket to expire objects after a retention period.
events come from handlers, webhooks, scheduled jobs - anywhere. a static emitter avoids threading an Application reference through every layer of your code. EventPublisher.emit() fires locally, register() catches it and publishes to sns.
register() controls what leaves the service. on() controls what enters. the boundary is explicit; consume without publishing, publish without consuming, or both.
errors in handlers are isolated, one failing handler doesn't block the rest of the batch. the failed message stays in sqs for retry.
class Application<Events extends Record<string, unknown>> {
constructor(config: ApplicationConfig)
register(event: string): void
on<PayloadType>(event: string, handler: (payload: PayloadType) => void | Promise<void>): void
on<Event extends keyof Events>(event: Event, handler: (payload: Events[Event]) => void | Promise<void>): void
start(): Promise<void>
stop(): Promise<void>
}type ApplicationConfig = {
name: string
environment: 'development' | 'staging' | 'production'
topic?: {
arn: string
config: SNSClientConfig
params?: Omit<PublishCommandInput, 'TopicArn' | 'Message'>
}
queue?: {
url: string
config: SQSClientConfig
params?: Omit<ReceiveMessageCommandInput, 'QueueUrl'>
}
storage?: {
bucket: string
config: S3ClientConfig
}
scheduler?: {
interval: number
}
}all service configs are optional, configure only what your service needs.
class DomainEvent<PayloadType> {
readonly id: string // uuid v7
readonly name: string // event name
readonly payload: PayloadType
readonly timestamp: number // epoch ms
}extend DomainEvent to define your events. use an intermediate base class to inject per-service metadata (app, category) once.
class EventPublisher {
static emit<EventType extends DomainEvent>(event: EventType): void
}emits a domain event to local listeners registered via Application.register(). can be called from anywhere in your process.