Give your ZenStack backend realtime streaming capabilities. Instantly react to any insert, update, or delete, and declaratively filter records using the same Prisma API you've come to love.
ℹ️ This project is not affiliated with or endorsed by the ZenStack team.
- 🛟 Type-safe: Queries have full intellisense based on your ZenStack models.
- 🛡️ Durable: Server went down? Your app picks up right where it left off, and even detects all the changes that happened while it was offline.
- 📈 Scalable: Just add more instances.
- 🗿 Simple: Build event-driven apps with fewer headaches.
- ZenStack (version >=
3.10.0) - Node.js (version >=
22.0.0) - Postgres that supports logical replication (version >=
15) - Redis (version >=
7.2.0) - A non-serverless server that will process events. You can forward events to serverless if you so desire.
npm install @visualbravo/zenstack-live
pnpm add @visualbravo/zenstack-live
bun add @visualbravo/zenstack-liveJust copy our docker-compose.yaml, make any necessary changes, and proceed as follows
import { schema } from './zenstack/schema'
import { ZenStackClient } from '@zenstackhq/orm'
import { ZenStackLive } from '@visualbravo/zenstack-live'
const client = new ZenStackClient(schema, {
...
})
const live = new ZenStackLive({
client,
redis: {
url: process.env.REDIS_URL,
},
})const stream = live.stream({
model: 'User',
slug: 'send-welcome-email',
when: {
created: {},
},
})
for await (const event of stream) {
const user = event.created
await sendEmail({
to: user.email,
subject: `Welcome, ${user.name}!`,
})
}Leaving out when, or leaving it empty, listens for all events.
import { beforeAfter } from '@visualbravo/zenstack-live'
const stream = live.stream({
model: 'Patient',
slug: 'audit-log',
})
for await (const event of stream) {
const { before, after } = beforeAfter(event)
await client.auditLog.create({
data: {
before,
after,
model: 'Patient',
operation: event.type,
},
})
}const stream = live.stream({
model: 'Post',
slug: 'cache',
})
for await (const event of stream) {
if (event.type === 'deleted') {
await redis.del(`posts:${event.deleted.id}`)
continue
}
const { after } = beforeAfter(event)
await redis.set(`posts:${after.id}`, after)
}You've seen the simple stuff, now let's look at more complex examples.
const stream = live.stream({
model: 'Post',
slug: 'sentiment-moderation',
when: {
created: {
OR: [
{
title: {
contains: 'ugly',
mode: 'insensitive',
},
},
{
title: {
contains: 'stupid',
mode: 'insensitive',
},
},
],
},
},
})
for await (const event of stream) {
const post = event.created
const sentiment = await analyzeSentiment(post)
if (sentiment === 'rude') {
await suspendUser(post.authorId)
}
}const stream = live.stream({
model: 'Order',
slug: 'send-delivered-email',
when: {
updated: {
after: {
status: 'DELIVERED',
},
},
},
})
for await (const event of stream) {
const order = event.updated.after
await sendEmail({
to: order.contactEmail,
subject: `✅ Delivered at ${toHumanReadable(event.date)}`,
})
}- LIVE adds the table of each stream to a publication, and Debezium connects to your database.
- Debezium stores inserts, updates, and deletes in a Redis stream, even those which are done outside of ZenStack.
- LIVE connects to Redis and reads the events in the stream.
- LIVE compares each event against your query, and if it matches, serves it to you.
MIT