Packages@g14o/events
Streams
Pluggable EventStream backends — memory, Redis, and Upstash.
Realtime delivery uses an EventStream: durable history + live fan-out. Same role as Redis Streams in Upstash Realtime.
Choosing memory vs Upstash vs Redis: Deployment.
Interface
Every adapter implements:
write(channel, message) → cursorreadAfter(channel, cursor)subscribe(channels, onMessage)close()
Pass a stream into new Event({ stream }). Omitting stream throws "Stream not configured".
Memory
import { memoryStream } from "@g14o/events/memory";
const event = new Event({
schema,
stream: memoryStream({ maxLength: 100 }),
});In-process only — fine for single-instance and local dev.
Redis (ioredis / node-redis)
import { Event } from "@g14o/events";
import { redisStream } from "@g14o/events/redis";
import Redis from "ioredis";
const event = new Event({
schema,
stream: redisStream({
client: new Redis(process.env.REDIS_URL!),
maxLength: 1000,
}),
});Upstash
import { Event } from "@g14o/events";
import { upstashStream } from "@g14o/events/upstash";
import { Redis } from "@upstash/redis";
const event = new Event({
schema,
stream: upstashStream({
redis: Redis.fromEnv(),
maxLength: 1000,
}),
});Default channelPrefix is @g14o:events. Do not put / in the prefix (or in channel names you pass through to Redis pub/sub keys): Upstash REST SUBSCRIBE places the channel in the URL path, so a slash splits the path and live fan-out fails while XADD / history still work.
Custom streams
import { defineStream } from "@g14o/events/stream";
const stream = defineStream({
async write(channel, message) {
/* persist + fan-out */
const id = "0-1";
return id;
},
async readAfter(channel, cursor) { return []; },
subscribe(channels, onMessage) { return () => {}; },
async close() {},
});