Eventa

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

// Define a typed event (the generic is the payload type)
const move = defineEventa<{ x: number, y: number }>()

// Create a base context (in-memory, useful for same-process communication)
const ctx = createContext()

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

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

const ctx = createContext()

// defineInvokeEventa<ResponseType, RequestType>(optionalName)
const echo = defineInvokeEventa<{ output: string }, { input: string }>('rpc:echo')

// Register handler (server side)
defineInvokeHandler(ctx, echo, ({ input }) => ({ output: input.toUpperCase() }))

// Create invoke function (client side)
const invokeEcho = defineInvoke(ctx, echo)
const result = await invokeEcho({ input: 'hello' }) // { output: 'HELLO' }

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

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

// Generator-style handler
defineStreamInvokeHandler(ctx, sync, async function* ({ jobId }) {
  for (let i = 1; i <= 5; i++) {
    yield { type: 'progress' as const, value: i * 20 }
  }
  yield { type: 'result' as const, value: 100 }
})

// Or imperative style with toStreamHandler
defineStreamInvokeHandler(ctx, sync, toStreamHandler(async ({ payload, emit }) => {
  emit({ type: 'progress', value: 0 })
  emit({ type: 'result', value: 100 })
}))

// Consume as async iterator
const stream = defineStreamInvoke(ctx, sync)
for await (const update of stream({ jobId: 'import' })) {
  console.log(update.type, update.value)
}

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 invoke = defineInvoke(ctx, recordRoute)
const input = new ReadableStream({
  start(c) { c.enqueue({ lat: 0, lng: 0 }); c.enqueue({ lat: 1, lng: 1 }); c.close() },
})
await invoke(input)

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 stream = defineStreamInvoke(ctx, routeChat)
for await (const note of stream(outgoing)) {
  console.log(note.message)
}

// Client-side cancellation
const controller = new AbortController()
const promise = invokeMethod({ input: 'work' }, { signal: controller.signal })
controller.abort('user cancelled')

// Server-side abort awareness
defineInvokeHandler(ctx, event, async ({ input }, options) => {
  const signal = options?.abortController?.signal
  if (signal?.aborted) return { output: 'aborted' }
  signal?.addEventListener('abort', () => { /* cleanup */ }, { once: true })
  return { output: `done: ${input}` }
})

import { linkChannel, pipeChannel } from '@moeru/eventa'

pipeChannel(a, b, c) // a -> b -> c
linkChannel(a, b, c) // a <-> b <-> c

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

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

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

import { createContext } from '@moeru/eventa/adapters/<adapter-name>'
const { context } = createContext(transportInstance)