bewinxed/river.ts/river-ts

import { RiverEvents } from 'river.ts';
const events = new RiverEvents()
  .defineEvent('message', { message: 'Hello' })
  .defineEvent('data', { data: {} as { id: number; name: string } })
  .defineEvent('stream', { data: [] as string[], stream: true, chunkSize: 100 })
  // Request/response pattern with explicit response type
  .defineEvent('rpc.call', {
    data: {} as { method: string; params: unknown },
    response: {} as { result: unknown; error?: string }
  })
  // Runtime validation: `data` is inferred from the schema's output
  .defineEvent('job.run', {
    schema: z.object({ id: z.string(), priority: z.number() }),
    responseSchema: z.object({ ok: z.boolean() }) // optional, for request()
  })
  .build();

import { RiverEmitter } from 'river.ts/server';
const emitter = RiverEmitter.init(events);
// Create SSE stream for HTTP response
const stream = emitter.stream({
  callback: async (emit, clientId) => {
    await emit('message', { message: 'Connected' });
    await emit('data', { data: { id: 1, name: 'test' } });
  },
  clientId: 'optional-custom-id',
  ondisconnect: (clientId) => console.log(`${clientId} disconnected`)
});
return new Response(stream, { headers: emitter.headers() });
// Broadcast to all clients
await emitter.broadcast('message', { message: 'Update' });
// Send to specific client
await emitter.sendToClient('client-id', 'data', { data: { id: 2, name: 'specific' } });

const stream = emitter.stream({
  retry: 3000,            // ms; written once as `retry:` when the stream opens
  keepAlive: 15_000,      // ms; writes a `: keep-alive` comment so proxies keep the connection
  lastEventId: request.headers.get('Last-Event-ID'), // read it from the request yourself
  signal: request.signal,
  callback: async (emit, clientId, lastEventId) => {
    // lastEventId is undefined on a first connection; replay everything after it
    for (const entry of log.after(lastEventId)) {
      await emit('message', { message: entry.text }, entry.id); // 3rd arg = event id
    }
  }
});
await emitter.broadcast('message', { message: 'Update' }, 42);          // optional id
await emitter.sendToClient('client-id', 'message', { message: 'x' }, 43); // optional id

import { RiverClient } from 'river.ts/client';
const client = RiverClient.init(events, { reconnect: true });
client
  .prepare('http://localhost:3000/events', { method: 'GET' })
  .on('message', (data) => console.log(data.message))
  .on('data', (data) => console.log(data.id, data.name))
  .stream();
// Close connection
client.close();
// stream() can be called again after close()

const client = RiverClient.init(events, {
  reconnect: true,                       // or { initialDelay: 1000, maxDelay: 30_000 }
  lastEventId: savedId,                  // sent as Last-Event-ID on the first connection
  onInvalid: (type, issues, raw) => {},  // events that fail their schema; never dispatched
  fetchFn: fetch,
  headers: { Authorization: 'Bearer ...' }
});
client.lastEventId; // read-only: id of the last event received; persist it to resume later
client.addEventListener('open', () => {});       // each successful connection
client.addEventListener('reconnect', (e) => {}); // (e as CustomEvent).detail = { attempt, delay, error }
client.addEventListener('close', () => {});      // stopped for good

import { RiverSocketAdapter } from 'river.ts/websocket';
const adapter = new RiverSocketAdapter(events, { debug: false });
// Register event handlers
adapter.on('message', (data) => console.log(data));
adapter.off('message', handler); // Unregister
// Handle incoming messages (call from ws.onmessage)
adapter.handleMessage(messageData);
// Send messages
adapter.send('data', { data: { id: 1, name: 'test' } }, (msg) => ws.send(msg));

import { RiverSocketAdapter, InvalidMessageError } from 'river.ts/websocket';
const adapter = new RiverSocketAdapter(events, {
  onInvalid: (type, issues, raw) => console.warn(type, issues, raw)
});

import {
  RiverSocketAdapter,
  RequestTimeoutError,
  WebSocketClosedError
} from 'river.ts/websocket';
// Events with explicit response types
const events = new RiverEvents()
  .defineEvent('instance.spawn', {
    data: {} as { cwd: string },
    response: {} as { instanceId: string; status: 'created' | 'error' }
  })
  .build();
const adapter = new RiverSocketAdapter(events);
// Route messages through adapter
ws.onmessage = (e) => adapter.handleMessage(e.data);
ws.onclose = () => adapter.clearPendingRequests();
// Make request - response type is inferred from event definition
const response = await adapter.request(
  'instance.spawn',
  { cwd: '/app' },
  (msg) => ws.send(msg),
  10000 // timeout in ms (default: 30000)
);
// response is typed as { instanceId: string; status: 'created' | 'error' }

// Request (outgoing)
{ "type": "instance.spawn", "data": { "cwd": "/app" }, "id": "uuid" }
// Response (incoming) - server echoes back the id
{ "type": "instance.spawn", "data": { "instanceId": "123", "status": "created" }, "id": "uuid" }

import { EventData, ResponseData, EmitPayload } from 'river.ts';
// EventData<T, K> - Extract data type for receiving/handling
// ResponseData<T, K> - Extract response type for request() return value
// EmitPayload<T, K> - Extract payload type for emitting (excludes type/stream/chunkSize/schema/responseSchema)
// InvalidHandler - (type, issues, raw) => void, the `onInvalid` signature
// InvalidMessageError - thrown by request() for a response that fails its schema

src/
├── index.ts          # Main exports (RiverEvents, types)
├── builder.ts        # RiverEvents builder class
├── validate.ts       # Standard Schema validation shared by client and websocket
├── client/           # RiverClient for SSE consumption
├── server/           # RiverEmitter for SSE streaming
├── websocket/        # RiverSocketAdapter for WebSocket
└── types/
    ├── core.ts       # BaseEvent, EventMap, EventData, ResponseData
    └── http.ts       # HTTPMethods type