eventa

Transport-aware events powering ergonomic RPC and streaming flows.

Heavily inspired by pragmatic RPC flows, but centred on pure events so transports stay swappable.

Warning

Eventa forwards whatever payload you emit. Validate data at the edges before sending it to untrusted peers.

Installation

npm install @moeru/eventa
pnpm i @moeru/eventa
bun i @moeru/eventa
ni @moeru/eventa
yarn add @moeru/eventa

Agent Skills

Install the eventa skill to your AI coding agent:

npx skills add moeru-ai/eventa

Getting Started

Event

It's very simple:

• defineEventa: all event should be defined with this util, it produces type safe constraints
• context: a channel bridges to peers (Electron, Worker, WebSocket Peer, you name it)
• createContext: to wrap any compatible event listener

If you need only events without RPC mechanism, then use with context.emit(...) and context.on(...)

import { createContext, defineEventa } from '@moeru/eventa'

const move = defineEventa<{ x: number, y: number }>()
const ctx = createContext()

ctx.emit(move, { x: 100, y: 200 })
ctx.on(move, ({ body }) => console.log(body.x, body.y))

RPC/Stream RPC

Events can be seen as packets transferring in networks, so we can use pure event to form a RPC/Stream RPC like how gRPC and tRPC works.

• defineInvokeEventa: define types of RPC/Stream RPC
• defineInvoke: this produce a function returns Promise for your RPC call to be used later, you can store and use it everywhere you want
• defineInvokeHandler: similar to how Nuxt, h3 defines their handler, we use defineInvokeHandler to hook a auto
• defineStreamInvokeHandler: similar to gRPC, when one RPC invocation produces not only one response, but multiple intermediate events, you may want to use it

Simple Example

The most simple way to show how it works:

import { createContext, defineInvoke, defineInvokeEventa, defineInvokeHandler } from '@moeru/eventa'

const ctx = createContext()
const someMethodDefine = defineInvokeEventa<{ output: string }, { input: number }>('random name')
defineInvokeHandler(ctx, someMethodDefine, ({ input }) => ({ output: String(input) }))

const someMethod = defineInvoke(ctx, someMethodDefine)
console.log(await someMethod(42)) // => { output: '42' }

Adapters

Eventa comes with various adapters for common use scenarios across browsers and Node.js, escalating the event orchestration in Electron, Web Workers, and WebSockets, etc.

Electron

1. Create a shared events module:

   import { defineInvokeEventa } from '@moeru/eventa'
   
   export const readdir = defineInvokeEventa<{ directories: string[] }, { cwd: string, target: string }>('rpc:node:fs/promise:readdir')

2. In the main process, bridge the adapter to ipcMain and your BrowserWindow instance:

   import { createContext as createMainContext } from '@moeru/eventa/adapters/electron/main'
   import { app, BrowserWindow, ipcMain } from 'electron'
   
   import { readdir } from './events/readdir'
   
   app.on('ready', () => {
     // ... other code
     const { context: mainCtx } = createMainContext(ipcMain, mainWindow.webContents)
     defineInvokeHandler(mainCtx, readdir, async ({ cwd, target }) => {
       const fs = await import('node:fs/promises')
       const path = await import('node:path')
       const fullPath = path.resolve(cwd, target)
       const directories = await fs.readdir(fullPath, { withFileTypes: true })
       return { directories: directories.filter(dirent => dirent.isDirectory()).map(dirent => dirent.name) }
     })
   })

3. In the renderer (not restricted to preload scripts, but recommended), bridge to ipcRenderer and expose a safe API:

   import { createContext as createRendererContext } from '@moeru/eventa/adapters/electron/renderer'
   import { contextBridge, ipcRenderer } from 'electron'
   
   import { defineInvoke, readdir } from './events/readdir'
   
   const { context: rendererCtx } = createRendererContext(ipcRenderer)
   const invokeReaddir = defineInvoke(rendererCtx, readdir)
   
   document.addEventListener('DOMContentLoaded', () => {
     invokeReaddir({ cwd: '/', target: 'usr' }).then((result) => {
       console.log('directories', result.directories)
     })
   })

4. The main and renderer contexts now share the invoke pipeline used throughout the examples in src/adapters/electron/*.test.ts.

Web Worker

1. Spawn the worker and wrap it with the main-thread adapter:

   import Worker from 'web-worker'
   
   import { createContext, defineInvoke, defineInvokeEventa } from '@moeru/eventa/adapters/webworkers'
   
   const worker = new Worker(new URL('./worker.js', import.meta.url), { type: 'module' })
   const { context: mainCtx } = createContext(worker)
   
   export const syncEvents = defineInvokeEventa<{ status: string }, { jobId: string }>('worker:sync')
   export const invokeSync = defineInvoke(mainCtx, syncEvents)

2. Inside the worker entry, create the worker context and register handlers:

   import { defineInvokeHandler } from '@moeru/eventa'
   import { createContext } from '@moeru/eventa/adapters/webworkers/worker'
   
   import { syncEvents } from '../sync'
   
   const { context: workerCtx } = createContext()
   defineInvokeHandler(workerCtx, syncEvents, ({ jobId }) => ({ status: `synced ${jobId}` }))

3. The same pattern works for streaming handlers and for sending transferrable(s) by switching to defineStreamInvoke or defineOutboundWorkerEventa as shown in src/adapters/webworkers/index.spec.ts.

BroadcastChannel

import { defineEventa } from '@moeru/eventa'
import { createContext } from '@moeru/eventa/adapters/broadcast-channel'

const channel = new BroadcastChannel('eventa-demo')
const { context: ctx } = createContext(channel)

const ping = defineEventa<{ message: string }>('bc:ping')
ctx.on(ping, ({ body }) => {
  console.log('received', body.message)
})

ctx.emit(ping, { message: 'Hello from BroadcastChannel' })

EventTarget

import { defineInvoke, defineInvokeEventa, defineInvokeHandler } from '@moeru/eventa'
import { createContext } from '@moeru/eventa/adapters/event-target'

const eventTarget = new EventTarget()
const { context: ctx } = createContext(eventTarget)

const echoEvents = defineInvokeEventa<{ output: string }, { input: string }>('et:echo')
defineInvokeHandler(ctx, echoEvents, ({ input }) => ({ output: input.toUpperCase() }))

const echo = defineInvoke(ctx, echoEvents)
console.log(await echo({ input: 'eventa' })) // => { output: 'EVENTA' }

EventEmitter (Node.js)

import { EventEmitter } from 'node:events'

import { defineEventa } from '@moeru/eventa'
import { createContext } from '@moeru/eventa/adapters/event-emitter'

const emitter = new EventEmitter()
const { context: ctx } = createContext(emitter)

const logEvent = defineEventa<{ message: string }>('emitter:log')
ctx.on(logEvent, ({ body }) => console.log(body.message))
ctx.emit(logEvent, { message: 'Hello from EventEmitter' })

Worker Threads (Node.js)

1. Main thread:

   import { Worker } from 'node:worker_threads'
   
   import { defineInvoke, defineInvokeEventa } from '@moeru/eventa'
   import { createContext } from '@moeru/eventa/adapters/worker-threads'
   
   const worker = new Worker(new URL('./worker.ts', import.meta.url), { type: 'module' })
   const { context: mainCtx } = createContext(worker)
   
   const helloEvents = defineInvokeEventa<{ output: string }, { input: string }>('node-worker-hello')
   const hello = defineInvoke(mainCtx, helloEvents)
   console.log(await hello({ input: 'Eventa' })) // => { output: 'Hello, Eventa' }

2. Worker entry:

   import { defineInvokeEventa, defineInvokeHandler } from '@moeru/eventa'
   import { createContext } from '@moeru/eventa/adapters/worker-threads/worker'
   
   const helloEvents = defineInvokeEventa<{ output: string }, { input: string }>('node-worker-hello')
   
   const { context: workerCtx } = createContext()
   defineInvokeHandler(workerCtx, helloEvents, ({ input }) => ({ output: `Hello, ${input}` }))

WebSocket (Client)

1. Open a WebSocket and wrap it with the native adapter:

   import { defineInvoke, defineInvokeEventa } from '@moeru/eventa'
   import { createContext as createWsContext } from '@moeru/eventa/adapters/websocket/native'
   
   const socket = new WebSocket('wss://example.com/ws')
   const { context: wsCtx } = createWsContext(socket)
   
   const chatEvents = defineInvokeEventa<{ message: string }, { text: string }>('chat:send')
   export const sendChat = defineInvoke(wsCtx, chatEvents)

2. Listen for connection lifecycle events to update UI state or retry logic:

   import { wsConnectedEvent, wsDisconnectedEvent } from '@moeru/eventa/adapters/websocket/native'
   
   wsCtx.on(wsConnectedEvent, () => console.log('connected'))
   wsCtx.on(wsDisconnectedEvent, () => console.log('disconnected'))

3. Pair the client with either the H3 global or peer adapter on the server for a full RPC channel (src/adapters/websocket/h3/*.test.ts).

WebSocket (Server with H3)

import { defineInvoke, defineInvokeHandler } from '@moeru/eventa'
// we support h3 by default, you can implement whatever you want, it's simple
import { createContext } from '@moeru/eventa/adapters/websocket/h3'

const chatEvents = defineInvokeEventa<{ message: string }, { text: string }>('chat:send')

const app = new H3()
const { untilLeastOneConnected, hooks } = createPeerHooks()
app.get('/ws', defineWebSocketHandler(hooks))

const { context } = await untilLeastOneConnected
defineInvokeHandler(context, chatEvents, ({ text: string }) => {
  // you can safely throw any error you want, you can even make the error type safe when using `defineInvoke`
  return { message: `Echo: ${text}` }
})

Advanced Usage

Streaming RPC

defineInvokeHandler is complemented by defineStreamInvokeHandler for long-running operations that need to report progress or intermediate results.

import { createContext, defineInvokeEventa, defineStreamInvoke, defineStreamInvokeHandler, toStreamHandler } from '@moeru/eventa'

const ctx = createContext()
const syncEvents = defineInvokeEventa<
  { type: 'progress' | 'result', value: number },
  { jobId: string }
>('rpc:sync')

// toStreamHelper converts an async function into an async generator
// so you can use imperative code instead of a generator function.
defineStreamInvokeHandler(ctx, syncEvents, toStreamHandler(async ({ payload, emit }) => {
  emit({ type: 'progress', value: 0 })
  for (let i = 1; i <= 5; i++) {
    emit({ type: 'progress', value: i * 20 })
  }
  emit({ type: 'result', value: 100 })
}))

const sync = defineStreamInvoke(ctx, syncEvents)
for await (const update of sync({ jobId: 'import' })) {
  console.log(update.type, update.value)
}

Both generator-style and imperative handlers are exercised in src/stream.spec.ts:7.

Abort/Cancel

Eventa supports cancellation via AbortSignal on the client side and exposes an AbortController inside handlers so you can stop work early.

Client-side (unary invoke):

import { createContext, defineInvoke, defineInvokeEventa } from '@moeru/eventa'

const ctx = createContext()
const slowEvents = defineInvokeEventa<{ output: string }, { input: string }>('rpc:slow')
const slow = defineInvoke(ctx, slowEvents)

const controller = new AbortController()
const promise = slow({ input: 'work' }, { signal: controller.signal })

controller.abort('user cancelled')
await promise // rejects with AbortError

Server-side handler (unary):

import { defineInvokeHandler } from '@moeru/eventa'

defineInvokeHandler(ctx, slowEvents, async ({ input }, options) => {
  const signal = options?.abortController?.signal
  if (signal?.aborted) {
    return { output: 'aborted' }
  }

  signal?.addEventListener('abort', () => {
    // clean up resources, cancel timers, close connections, etc.
  }, { once: true })

  // ... do work
  return { output: `done: ${input}` }
})

Client-side (stream invoke):

import { defineInvokeEventa, defineStreamInvoke } from '@moeru/eventa'

const streamEvents = defineInvokeEventa<{ type: 'progress' | 'done', value: number }, { jobId: string }>('rpc:stream')
const stream = defineStreamInvoke(ctx, streamEvents)

const controller = new AbortController()
const results = stream({ jobId: 'import' }, { signal: controller.signal })

setTimeout(() => controller.abort('timeout'), 1000)
for await (const msg of results) {
  console.log(msg)
}

Server-side handler (streaming):

import { defineStreamInvokeHandler } from '@moeru/eventa'

defineStreamInvokeHandler(ctx, streamEvents, async function* ({ jobId }, options) {
  const signal = options?.abortController?.signal

  for (let i = 0; i <= 5; i++) {
    if (signal?.aborted) {
      return
    }
    yield { type: 'progress', value: i * 20 }
    await new Promise(r => setTimeout(r, 200))
  }

  yield { type: 'done', value: 100 }
})

Streaming Input

Eventa supports stream inputs on unary invokes (client-streaming) and full bidirectional streaming. This mirrors the gRPC shapes:

// Client-streaming request -> unary response
rpc RecordRoute(stream Point) returns (RouteSummary) {}

// Bidirectional streaming request/response
rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

Client-streaming input with defineInvoke (stream in, single response out):

import { createContext, defineInvoke, defineInvokeEventa, defineInvokeHandler } from '@moeru/eventa'

const ctx = createContext()

const recordRoute = defineInvokeEventa<
  { distance: number, points: number },
  ReadableStream<{ lat: number, lng: number }>
>('rpc:record-route')

defineInvokeHandler(ctx, recordRoute, async (stream) => {
  let points = 0
  for await (const _ of stream) {
    points += 1
  }
  return { distance: points * 10, points }
})

const input = new ReadableStream({
  start(controller) {
    controller.enqueue({ lat: 0, lng: 0 })
    controller.enqueue({ lat: 1, lng: 1 })
    controller.close()
  },
})

const invoke = defineInvoke(ctx, recordRoute)
console.log(await invoke(input))

Bidirectional streaming with defineStreamInvoke (stream in, stream out):

import { createContext, defineInvokeEventa, defineStreamInvoke, defineStreamInvokeHandler } from '@moeru/eventa'

const ctx = createContext()

const routeChat = defineInvokeEventa<
  { message: string },
  ReadableStream<{ message: string }>
>('rpc:route-chat')

defineStreamInvokeHandler(ctx, routeChat, async function* (incoming) {
  for await (const note of incoming) {
    yield { message: `echo: ${note.message}` }
  }
})

const outgoing = new ReadableStream({
  start(controller) {
    controller.enqueue({ message: 'hello' })
    controller.enqueue({ message: 'from stream' })
    controller.close()
  },
})

const stream = defineStreamInvoke(ctx, routeChat)
for await (const note of stream(outgoing)) {
  console.log(note.message)
}

Shorthands for defineInvokeHandler and defineInvoke

When you have multiple invoke events to register handlers for, or to create invoke functions for, you can use defineInvokeHandlers and defineInvokes to do so in bulk.

const events = {
  double: defineInvokeEventa<number, number>(),
  append: defineInvokeEventa<string, string>(),
}

defineInvokeHandlers(ctx, events, {
  double: input => input * 2,
  append: input => `${input}!`,
})

const {
  double: invokeDouble,
  append: invokeAppend,
} = defineInvokes(ctx, events)

await invokeDouble(5) // 10
await invokeAppend('test') // 'test!'

Development

pnpm i
pnpm test

Note

pnpm test runs Vitest interactively. Use pnpm test:run for a single pass.

Similar projects

• birpc: We dislike the way the API designs, we want fully free sharable invok-able functions, streaming input, streaming output, etc.
• async-call-rpc: it only works with JSON-RPC, but the DX is similar

License

MIT