You need realtime updates — live notifications, collaborative editing, streaming data. So you add Socket.io for WebSockets or a custom SSE endpoint. Now you have two servers, two auth systems, and message types that are strings you parse with JSON.parse and hope for the best.
Same functions. Now realtime.
You write this
// Functions — same as your API
const getMessages = pikkuFunc({
input: z.object({ channelId: z.string() }),
output: z.array(MessageSchema),
func: async ({ db }, { channelId }) =>
db.messages.list({ channelId }),
permissions: { user: isChannelMember },
})
const sendMessage = pikkuFunc({
input: z.object({ channelId: z.string(), text: z.string() }),
output: MessageSchema,
func: async ({ db }, data) =>
db.messages.create(data),
permissions: { user: isChannelMember },
})
// Wire to WebSocket
wireChannel({
channel: 'chat',
route: '/chat',
onConnect: getMessages,
onMessage: { sendMessage },
})
// Wire to SSE (one-way streaming)
wireSSE({ route: '/messages/stream', func: getMessages })
// Same functions also work as HTTP
wireHTTP({ method: 'get', route: '/messages/:channelId', func: getMessages })
wireHTTP({ method: 'post', route: '/messages', func: sendMessage })
What Fabric gives you
Typed messages
Input and output schemas validated on both sides. No JSON.parse guessing.
Session auth
WebSocket connections authenticated with the same session as your HTTP API.
Observable
Connection counts, message rates, errors — same dashboard as everything else.
client (auto-generated)
// Type-safe WebSocket client
const channel = client.chat.connect()
channel.on('sendMessage', (msg) => {
// msg is typed as Message
})
Realtime without a second codebase.
Same functions. Now over WebSocket. Deploy.