Add per-adapter architecture docs in event-targets/ directory
- Create docs/architecture/event-targets/ with individual specs: in-process, redis, websocket-client, websocket-server, worker, iroh-spoke, iroh-hub - Update event-targets.md to serve as index with topology model (symmetric vs fan-out) and adapter status table - Update architecture.md index to reference new directory
This commit is contained in:
@@ -1,11 +1,11 @@
|
||||
---
|
||||
status: draft
|
||||
last_updated: 2026-05-01
|
||||
last_updated: 2026-05-07
|
||||
---
|
||||
|
||||
# Event Target Adapters
|
||||
|
||||
In-process, Redis, WebSocket, and Worker event targets. All implement `TypedEventTarget<TEvent>`.
|
||||
All adapters implement the `TypedEventTarget<TEvent>` interface and use the `EventEnvelope` format (`{ type, id, payload }`) as the serialization contract.
|
||||
|
||||
## Interface Contract
|
||||
|
||||
@@ -17,139 +17,23 @@ Every adapter must implement:
|
||||
| `dispatchEvent(event)` | Send/dispatch event. Returns `boolean` (always `true` for non-cancelable events). |
|
||||
| `removeEventListener(type, callback)` | Unregister listener. Clean up underlying subscription when no listeners remain for a topic. |
|
||||
|
||||
All adapters use the `EventEnvelope` format (`{ type, id, payload }`) as the serialization contract. Adapters that cross process boundaries (Redis, WebSocket, Iroh) serialize/deserialize the full envelope as JSON.
|
||||
## Topology Model
|
||||
|
||||
## In-Process (Default)
|
||||
Adapters come in two shapes:
|
||||
|
||||
No adapter needed. `createPubSub` uses `new EventTarget()` by default. This works for single-process deployments where all pubsub participants share the same memory.
|
||||
- **Symmetric** (single connection) — wraps one connection. Same interface on both sides. Examples: Redis, Iroh spoke, WebSocket client, Worker main-thread.
|
||||
- **Fan-out** (multi-connection) — manages multiple connections. `dispatchEvent` sends to all; `addEventListener` aggregates from all. Examples: WebSocket server, Iroh hub, Worker pool manager.
|
||||
|
||||
No explicit `InProcessEventTarget` class — the web standard `EventTarget` already implements the interface. Could be formalized later if a name makes the API clearer, but `new EventTarget()` is already the standard.
|
||||
The `createPubSub` layer is topology-agnostic. A hub composes multiple adapters and uses operators to combine streams — this is downstream application logic, not a package boundary.
|
||||
|
||||
## Redis
|
||||
## Adapter Docs
|
||||
|
||||
**Import**: `@alkdev/pubsub/event-target-redis`
|
||||
**Peer dep**: `ioredis@^5.0.0` (optional)
|
||||
|
||||
### `createRedisEventTarget`
|
||||
|
||||
```ts
|
||||
function createRedisEventTarget<TEvent extends TypedEvent>(
|
||||
args: CreateRedisEventTargetArgs,
|
||||
): TypedEventTarget<TEvent>;
|
||||
```
|
||||
|
||||
### `CreateRedisEventTargetArgs`
|
||||
|
||||
| Field | Type | Required | Description |
|
||||
|-------|------|----------|-------------|
|
||||
| `publishClient` | `Redis \| Cluster` | Yes | ioredis client for publishing. Can share a connection. |
|
||||
| `subscribeClient` | `Redis \| Cluster` | Yes | ioredis client for subscribing. Must be dedicated — Redis requires subscriber connections to only receive messages. |
|
||||
| `serializer` | `{ stringify, parse }` | No | Custom serializer. Defaults to `JSON`. |
|
||||
|
||||
### How It Works
|
||||
|
||||
- `dispatchEvent` → `publishClient.publish(event.type, serializer.stringify(event.detail))`
|
||||
- `addEventListener` → `subscribeClient.subscribe(topic)`, track callbacks per topic
|
||||
- `removeEventListener` → remove callback; if no callbacks remain for topic, `subscribeClient.unsubscribe(topic)`
|
||||
- On message: deserializes with `serializer.parse`, reconstructs `CustomEvent(channel, { detail: envelope })`
|
||||
|
||||
The `detail` of the `CustomEvent` dispatched to local listeners is the full `EventEnvelope` object (`{ type, id, payload }`).
|
||||
|
||||
### Channel Naming
|
||||
|
||||
Currently uses the topic string directly as the Redis channel name (e.g., `session.status:proj_123`). Architecture recommends `alk:events:{eventType}` prefix but this is not yet implemented. Should be configurable: `createRedisEventTarget({ ..., prefix: "alk:events:" })`.
|
||||
|
||||
### Limitations (Current)
|
||||
|
||||
- **No error handling** — connection failures, reconnection, and message parse errors are not handled
|
||||
- **No channel prefix** — raw event types as channel names risk collision in shared Redis instances
|
||||
- **No unsubscribe cleanup on client disconnect** — if the subscribe client disconnects, registered callbacks remain in the map but will never fire
|
||||
|
||||
### Test Coverage
|
||||
|
||||
No tests yet (test directory is empty). Previous alkhub had 5 Redis tests (publish path only, mocked ioredis).
|
||||
|
||||
## WebSocket
|
||||
|
||||
**Import**: `@alkdev/pubsub/event-target-websocket` (not yet implemented)
|
||||
**Peer dep**: none (WebSocket is a web standard)
|
||||
|
||||
### Design (Spec from `spoke-runner.md`)
|
||||
|
||||
```ts
|
||||
class WebSocketEventTarget implements TypedEventTarget<any> {
|
||||
private listeners = new Map<string, Set<(event: CustomEvent) => void>>()
|
||||
|
||||
constructor(private ws: WebSocket) {
|
||||
ws.onmessage = (msg) => {
|
||||
const envelope = JSON.parse(msg.data as string) // { type, id, payload }
|
||||
const topic = `${envelope.type}:${envelope.id}`
|
||||
const event = new CustomEvent(topic, { detail: envelope })
|
||||
for (const listener of this.listeners.get(topic) ?? []) {
|
||||
listener(event)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
dispatchEvent(event: CustomEvent): boolean {
|
||||
this.ws.send(JSON.stringify(event.detail)) // sends { type, id, payload }
|
||||
return true
|
||||
}
|
||||
|
||||
addEventListener(type: string, listener: (event: CustomEvent) => void): void { ... }
|
||||
removeEventListener(type: string, listener: (event: CustomEvent) => void): void { ... }
|
||||
}
|
||||
```
|
||||
|
||||
### Key Properties
|
||||
|
||||
- **Bidirectional** — `dispatchEvent` sends over WS, `addEventListener` receives from WS
|
||||
- **Per-connection** — hub creates one per spoke connection
|
||||
- **JSON framing** — WebSocket provides native message boundaries (no length-prefix needed)
|
||||
- **No native deps** — works in browsers and Node
|
||||
- **Envelope serialization** — sends/receives the full `EventEnvelope` JSON (`{ type, id, payload }`)
|
||||
|
||||
### Gap: Reconnection
|
||||
|
||||
WebSocket connections drop. On reconnect, the spoke must re-register with the hub (same `hub.register` flow). The `WebSocketEventTarget` itself is per-connection — a new connection means a new event target instance. Reconnection logic belongs to the spoke lifecycle, not the event target.
|
||||
|
||||
## Worker
|
||||
|
||||
**Import**: `@alkdev/pubsub/event-target-worker` (not yet implemented)
|
||||
**Peer dep**: none (Web Worker API is standard)
|
||||
|
||||
### Design
|
||||
|
||||
A `WorkerEventTarget` implementing `TypedEventTarget` over `postMessage`/`onmessage`. This enables `createPubSub` to work across Web Worker boundaries.
|
||||
|
||||
The worker message protocol uses the `EventEnvelope` format:
|
||||
|
||||
```json
|
||||
{ "type": "call.responded", "id": "uuid-123", "payload": { "output": 42 } }
|
||||
```
|
||||
|
||||
### Two-Sided Design
|
||||
|
||||
- **Main thread** (`WorkerPoolManager` side): dispatches typed messages to workers via `worker.postMessage()`, receives responses via `worker.onmessage`
|
||||
- **Worker thread**: dispatches to main thread via `parentPort.postMessage()`, receives from main thread via `globalThis.onmessage`
|
||||
|
||||
Both sides wrap `postMessage`/`onmessage` to implement the `TypedEventTarget` interface:
|
||||
|
||||
```ts
|
||||
// Main thread side
|
||||
const workerEventTarget = createWorkerEventTarget(worker);
|
||||
|
||||
// Worker thread side
|
||||
const mainEventTarget = createMainThreadEventTarget();
|
||||
```
|
||||
|
||||
### Key Properties
|
||||
|
||||
- **Bidirectional** — both sides can publish and subscribe
|
||||
- **Per-worker** — each worker gets its own event target
|
||||
- **Structured clone** — Web Workers use structured clone algorithm for serialization, but JSON-serializable `EventEnvelope` ensures cross-platform compatibility
|
||||
- **No native deps** — works in any environment with Web Worker support
|
||||
|
||||
### Relationship to Taskgraph / Operations
|
||||
|
||||
The worker event target enables distributed operation execution. Workers can subscribe to `call.requested` events and publish `call.responded` events through the event target, allowing `@alkdev/operations` to dispatch work to worker threads via the same pubsub transport.
|
||||
| Adapter | Import | Status |
|
||||
|---------|--------|--------|
|
||||
| [In-Process](in-process.md) | (default, no import) | Implemented (built-in `EventTarget`) |
|
||||
| [Redis](redis.md) | `@alkdev/pubsub/event-target-redis` | Implemented. Needs tests. |
|
||||
| [WebSocket Client](websocket-client.md) | `@alkdev/pubsub/event-target-websocket-client` | Not yet implemented |
|
||||
| [WebSocket Server](websocket-server.md) | `@alkdev/pubsub/event-target-websocket-server` | Not yet implemented |
|
||||
| [Worker](worker.md) | `@alkdev/pubsub/event-target-worker` | Not yet implemented (R&D on Node vs Web Worker) |
|
||||
| [Iroh Spoke](iroh-spoke.md) | `@alkdev/pubsub/event-target-iroh-spoke` | Not yet implemented (R&D on binding) |
|
||||
| [Iroh Hub](iroh-hub.md) | `@alkdev/pubsub/event-target-iroh-hub` | Not yet implemented (R&D on binding) |
|
||||
Reference in New Issue
Block a user