Implement event-driven API architecture
Generates production-ready event-driven APIs with pub/sub patterns, message queues, and async communication for microservices.
/plugin marketplace add jeremylongshore/claude-code-plugins-plus-skills/plugin install api-fuzzer@claude-code-plugins-plusBuild production-grade event-driven APIs with message queues, event streaming, and async communication patterns. This command generates event publishers, subscribers, message brokers integration, and event-driven architectures for microservices and distributed systems.
Why event-driven architecture:
Alternatives considered:
This approach balances: Loose coupling, reliability, scalability, and operational complexity.
Use event-driven architecture when:
Don't use when:
Choose Message Broker
Define Event Schemas
Implement Publishers
Build Subscribers
Add Event Patterns
// events/EventPublisher.js
const amqp = require('amqplib');
const { v4: uuidv4 } = require('uuid');
class EventPublisher {
constructor(connectionUrl) {
this.connectionUrl = connectionUrl;
this.connection = null;
this.channel = null;
}
async connect() {
this.connection = await amqp.connect(this.connectionUrl);
this.channel = await this.connection.createChannel();
// Declare exchange for fanout (pub/sub)
await this.channel.assertExchange('events', 'topic', { durable: true });
console.log('Event publisher connected to RabbitMQ');
}
async publish(eventName, payload) {
if (!this.channel) {
throw new Error('Publisher not connected');
}
const event = {
id: uuidv4(),
name: eventName,
timestamp: new Date().toISOString(),
version: '1.0.0',
data: payload
};
const routingKey = eventName;
const message = Buffer.from(JSON.stringify(event));
const published = this.channel.publish(
'events',
routingKey,
message,
{
persistent: true, // Survive broker restart
contentType: 'application/json',
messageId: event.id,
timestamp: Date.now()
}
);
if (!published) {
throw new Error('Failed to publish event to exchange buffer');
}
console.log(`Published event: ${eventName}`, event.id);
return event.id;
}
async close() {
await this.channel?.close();
await this.connection?.close();
}
}
module.exports = EventPublisher;
// Usage in API route
const publisher = new EventPublisher('amqp://localhost');
await publisher.connect();
router.post('/users', async (req, res) => {
try {
const user = await createUser(req.body);
// Publish event after successful creation
await publisher.publish('user.created', {
userId: user.id,
email: user.email,
name: user.name
});
res.status(201).json(user);
} catch (error) {
next(error);
}
});
// events/EventSubscriber.js
const amqp = require('amqplib');
class EventSubscriber {
constructor(connectionUrl, queueName) {
this.connectionUrl = connectionUrl;
this.queueName = queueName;
this.handlers = new Map();
}
async connect() {
this.connection = await amqp.connect(this.connectionUrl);
this.channel = await this.connection.createChannel();
// Declare exchange
await this.channel.assertExchange('events', 'topic', { durable: true });
// Declare queue with dead-letter exchange
await this.channel.assertQueue(this.queueName, {
durable: true,
deadLetterExchange: 'events.dlx',
deadLetterRoutingKey: 'dead-letter'
});
console.log(`Subscriber ${this.queueName} connected`);
}
on(eventName, handler) {
this.handlers.set(eventName, handler);
// Bind queue to routing key
this.channel.bindQueue(this.queueName, 'events', eventName);
console.log(`Subscribed to event: ${eventName}`);
}
async start() {
this.channel.prefetch(1); // Process one message at a time
this.channel.consume(this.queueName, async (message) => {
if (!message) return;
const content = message.content.toString();
const event = JSON.parse(content);
console.log(`Received event: ${event.name}`, event.id);
const handler = this.handlers.get(event.name);
if (!handler) {
console.warn(`No handler for event: ${event.name}`);
this.channel.ack(message); // Acknowledge to prevent reprocessing
return;
}
try {
// Idempotency: Check if already processed
const alreadyProcessed = await checkEventProcessed(event.id);
if (alreadyProcessed) {
console.log(`Event already processed: ${event.id}`);
this.channel.ack(message);
return;
}
// Process event
await handler(event.data, event);
// Mark as processed
await markEventProcessed(event.id);
// Acknowledge successful processing
this.channel.ack(message);
} catch (error) {
console.error(`Error processing event ${event.name}:`, error);
// Reject and requeue (will go to DLQ after max retries)
this.channel.nack(message, false, false);
}
});
console.log(`Subscriber ${this.queueName} started`);
}
}
// Usage
const subscriber = new EventSubscriber('amqp://localhost', 'email-service');
await subscriber.connect();
subscriber.on('user.created', async (data, event) => {
await sendWelcomeEmail(data.email, data.name);
console.log(`Welcome email sent to ${data.email}`);
});
subscriber.on('order.placed', async (data, event) => {
await sendOrderConfirmation(data.orderId, data.email);
console.log(`Order confirmation sent for ${data.orderId}`);
});
await subscriber.start();
# events/kafka_producer.py
from kafka import KafkaProducer
from kafka.errors import KafkaError
import json
import uuid
from datetime import datetime
from typing import Dict, Any
class EventProducer:
def __init__(self, bootstrap_servers: str):
self.producer = KafkaProducer(
bootstrap_servers=bootstrap_servers,
value_serializer=lambda v: json.dumps(v).encode('utf-8'),
acks='all', # Wait for all replicas
retries=3,
max_in_flight_requests_per_connection=1 # Preserve order
)
def publish(self, topic: str, event_name: str, payload: Dict[str, Any]) -> str:
event_id = str(uuid.uuid4())
event = {
'id': event_id,
'name': event_name,
'timestamp': datetime.utcnow().isoformat(),
'version': '1.0.0',
'data': payload
}
future = self.producer.send(
topic,
value=event,
key=event_id.encode('utf-8') # Partition by event ID
)
try:
# Block for synchronous send (optional)
record_metadata = future.get(timeout=10)
print(f"Published {event_name} to {record_metadata.topic} "
f"partition {record_metadata.partition} offset {record_metadata.offset}")
return event_id
except KafkaError as e:
print(f"Failed to publish event: {e}")
raise
def close(self):
self.producer.flush()
self.producer.close()
# Usage
producer = EventProducer('localhost:9092')
def create_user_endpoint(request):
user = create_user(request.data)
# Publish event
producer.publish(
topic='user-events',
event_name='user.created',
payload={
'user_id': user.id,
'email': user.email,
'name': user.name
}
)
return {'user': user.to_dict()}, 201
// Event sourcing: Store events as source of truth
class OrderEventStore {
constructor(publisher) {
this.publisher = publisher;
}
async placeOrder(orderId, items, customerId) {
// Publish event (this is the source of truth)
await this.publisher.publish('order.placed', {
orderId,
items,
customerId,
status: 'pending',
timestamp: new Date().toISOString()
});
}
async confirmPayment(orderId, paymentId) {
await this.publisher.publish('order.payment_confirmed', {
orderId,
paymentId,
timestamp: new Date().toISOString()
});
}
async shipOrder(orderId, trackingNumber) {
await this.publisher.publish('order.shipped', {
orderId,
trackingNumber,
timestamp: new Date().toISOString()
});
}
}
// Rebuild order state from events
async function getOrderState(orderId) {
const events = await loadEvents(`order.${orderId}.*`);
let order = { id: orderId };
for (const event of events) {
switch (event.name) {
case 'order.placed':
order = { ...order, ...event.data, status: 'pending' };
break;
case 'order.payment_confirmed':
order.status = 'confirmed';
order.paymentId = event.data.paymentId;
break;
case 'order.shipped':
order.status = 'shipped';
order.trackingNumber = event.data.trackingNumber;
break;
}
}
return order;
}
// Write model: Handle commands, emit events
class OrderCommandHandler {
async handlePlaceOrder(command) {
// Validate
if (!command.items.length) {
throw new Error('Order must have items');
}
// Create order (write)
const order = await db.orders.create({
customerId: command.customerId,
items: command.items,
status: 'pending'
});
// Emit event
await publisher.publish('order.placed', {
orderId: order.id,
customerId: order.customerId,
totalAmount: calculateTotal(order.items)
});
return order.id;
}
}
// Read model: Listen to events, update read-optimized views
subscriber.on('order.placed', async (data) => {
// Update denormalized view for fast queries
await redis.set(`order:${data.orderId}`, JSON.stringify({
orderId: data.orderId,
customerId: data.customerId,
totalAmount: data.totalAmount,
status: 'pending',
placedAt: new Date().toISOString()
}));
// Update customer order count
await redis.hincrby(`customer:${data.customerId}`, 'orderCount', 1);
});
// Orchestrate multi-service transaction with compensating actions
class OrderSaga {
async execute(orderData) {
const sagaId = uuidv4();
try {
// Step 1: Reserve inventory
await publisher.publish('inventory.reserve', {
sagaId,
items: orderData.items
});
await waitForEvent('inventory.reserved', sagaId);
// Step 2: Process payment
await publisher.publish('payment.process', {
sagaId,
amount: orderData.amount,
customerId: orderData.customerId
});
await waitForEvent('payment.processed', sagaId);
// Step 3: Create shipment
await publisher.publish('shipment.create', {
sagaId,
orderId: orderData.orderId,
address: orderData.shippingAddress
});
await waitForEvent('shipment.created', sagaId);
// Saga completed successfully
await publisher.publish('order.saga_completed', { sagaId });
} catch (error) {
// Compensate: Rollback in reverse order
console.error('Saga failed, executing compensations', error);
await publisher.publish('shipment.cancel', { sagaId });
await publisher.publish('payment.refund', { sagaId });
await publisher.publish('inventory.release', { sagaId });
await publisher.publish('order.saga_failed', { sagaId, reason: error.message });
}
}
}
Common issues and solutions:
Problem: Events lost during broker outage
Problem: Duplicate event processing
Problem: Events processed out of order
Problem: Subscriber can't keep up with events
Problem: Dead-letter queue fills up
Transactional outbox pattern (prevent lost events):
// Atomic database write + event publish
async function createUserWithEvent(userData) {
const transaction = await db.transaction();
try {
// 1. Create user in database
const user = await db.users.create(userData, { transaction });
// 2. Store event in outbox table (same transaction)
await db.outbox.create({
eventName: 'user.created',
payload: { userId: user.id, email: user.email },
published: false
}, { transaction });
await transaction.commit();
// 3. Background job publishes from outbox
// If app crashes, outbox worker retries unpublished events
} catch (error) {
await transaction.rollback();
throw error;
}
}
const userCreatedSchema = {
$schema: "http://json-schema.org/draft-07/schema#",
type: "object",
required: ["id", "name", "timestamp", "version", "data"],
properties: {
id: { type: "string", format: "uuid" },
name: { type: "string", const: "user.created" },
timestamp: { type: "string", format: "date-time" },
version: { type: "string", pattern: "^\\d+\\.\\d+\\.\\d+$" },
data: {
type: "object",
required: ["userId", "email"],
properties: {
userId: { type: "integer" },
email: { type: "string", format: "email" },
name: { type: "string", minLength: 1 }
}
}
}
};
const rabbitConfig = {
url: process.env.RABBITMQ_URL || 'amqp://localhost',
exchange: {
name: 'events',
type: 'topic', // Supports wildcard routing (user.*, order.created)
durable: true // Survive broker restart
},
queue: {
durable: true,
deadLetterExchange: 'events.dlx',
messageTtl: 86400000, // 24 hours
maxLength: 100000, // Max messages in queue
maxPriority: 10 // Priority queue support
},
publisher: {
confirm: true, // Wait for broker acknowledgment
persistent: true // Messages survive broker restart
},
subscriber: {
prefetch: 1, // Messages to prefetch
noAck: false, // Manual acknowledgment
exclusive: false // Allow multiple consumers
}
};
DO:
DON'T:
TIPS:
<entity>.<action> (user.created, order.shipped)/build-api-gateway - Route events through API gateway/generate-rest-api - Generate REST API that publishes events/create-monitoring - Monitor event processing metrics/implement-throttling - Rate limit event publishing/scan-api-security - Security scan event handlersOptimization strategies:
// Batch events for higher throughput
const eventBatch = [];
setInterval(async () => {
if (eventBatch.length > 0) {
await publisher.publishBatch(eventBatch);
eventBatch.length = 0;
}
}, 100); // Flush every 100ms
// Parallel event processing (if order doesn't matter)
subscriber.channel.prefetch(10); // Process 10 messages concurrently
Security checklist:
// Use TLS for RabbitMQ
const connection = await amqp.connect('amqps://user:pass@broker:5671', {
ca: [fs.readFileSync('ca-cert.pem')],
cert: fs.readFileSync('client-cert.pem'),
key: fs.readFileSync('client-key.pem')
});
// Validate event schemas
const Ajv = require('ajv');
const ajv = new Ajv();
const validate = ajv.compile(eventSchema);
function publishEvent(event) {
if (!validate(event)) {
throw new Error(`Invalid event: ${ajv.errorsText(validate.errors)}`);
}
// Proceed with publish
}
Events not being consumed:
High event processing lag:
Events being processed multiple times:
Dead-letter queue filling up: