diff --git a/backends/mizan-django/src/mizan/channels/connection.py b/backends/mizan-django/src/mizan/channels/connection.py index 2d2a59d..29f7bc4 100644 --- a/backends/mizan-django/src/mizan/channels/connection.py +++ b/backends/mizan-django/src/mizan/channels/connection.py @@ -18,7 +18,7 @@ Protocol: {"channel": "chat", "params": {"room": "general"}, "type": "DjangoMessage", "data": {...}} # RPC responses - {"id": "request-id", "ok": true, "data": {...}} + {"id": "request-id", "ok": true, "data": {"result": {...}, "invalidate": [...]}} {"id": "request-id", "ok": false, "error": {...}} {"error": "..."} @@ -390,7 +390,7 @@ class DjangoReactConsumer(AsyncJsonWebsocketConsumer): Protocol: Request: {"action": "rpc", "id": "request-id", "fn": "function_name", "args": {...}} - Response: {"id": "request-id", "ok": true, "data": {...}} + Response: {"id": "request-id", "ok": true, "data": {"result":..., "invalidate":[...]}} or: {"id": "request-id", "ok": false, "error": {...}} Security: @@ -485,13 +485,23 @@ class DjangoReactConsumer(AsyncJsonWebsocketConsumer): } ) else: - await self.send_json( - { - "id": request_id, - "ok": True, - "data": result.data, - } + # the same {result, invalidate, merge} envelope the HTTP RPC path builds, so a + # mutation sent over the socket invalidates exactly as one sent over HTTP + from mizan.client.executor import _resolve_invalidation, _resolve_merges + + data = {"result": result.data} + invalidate = await sync_to_async(_resolve_invalidation, thread_sensitive=True)( + fn_class, args ) + merges = await sync_to_async(_resolve_merges, thread_sensitive=True)( + fn_class, args, result.data + ) + if invalidate: + data["invalidate"] = invalidate + if merges: + data["merge"] = merges + + await self.send_json({"id": request_id, "ok": True, "data": data}) async def channel_message(self, event: dict): """ diff --git a/backends/mizan-django/src/mizan/tests/test_channels.py b/backends/mizan-django/src/mizan/tests/test_channels.py index 7f89f31..5d305b9 100644 --- a/backends/mizan-django/src/mizan/tests/test_channels.py +++ b/backends/mizan-django/src/mizan/tests/test_channels.py @@ -1000,7 +1000,8 @@ class WebSocketRPCTests(TestCase): self.assertEqual(response["id"], "test-123") self.assertTrue(response["ok"]) - self.assertEqual(response["data"]["echo"], "Echo: Hello") + # data is the {result, invalidate, merge} envelope, as on the HTTP RPC path + self.assertEqual(response["data"]["result"]["echo"], "Echo: Hello") def test_handle_rpc_with_multiple_args(self): """_handle_rpc should handle functions with multiple arguments.""" @@ -1029,7 +1030,7 @@ class WebSocketRPCTests(TestCase): response = consumer.sent_messages[0] self.assertTrue(response["ok"]) - self.assertEqual(response["data"]["result"], 8) + self.assertEqual(response["data"]["result"]["result"], 8) def test_handle_rpc_function_not_found(self): """_handle_rpc should return error for unknown function.""" diff --git a/backends/mizan-fastapi/src/mizan_fastapi/websocket.py b/backends/mizan-fastapi/src/mizan_fastapi/websocket.py index 412e860..3ff0f8b 100644 --- a/backends/mizan-fastapi/src/mizan_fastapi/websocket.py +++ b/backends/mizan-fastapi/src/mizan_fastapi/websocket.py @@ -6,6 +6,7 @@ The WebSocket endpoint — channel subscriptions and RPC over one connection. {"action": "unsubscribe", "channel": "chat", "params": {...}} {"action": "message", "channel": "chat", "params": {...}, "data": {...}} {"action": "rpc", "id": "request-id", "fn": "function_name", "args": {...}} + {"action": "ctx", "id": "request-id", "context": "name", "params": {...}} Server sends: {"channel": "chat", "params": {...}, "type": "...", "data": {...}} @@ -29,7 +30,7 @@ from typing import Any from fastapi import APIRouter, WebSocket, WebSocketDisconnect -from mizan_core.registry import get_function +from mizan_core.registry import get_context_groups, get_function from mizan_fastapi import channels from mizan_fastapi.executor import ( MizanError, @@ -162,11 +163,41 @@ async def _rpc(socket: WebSocket, body: dict[str, Any]) -> None: await socket.send_json({"id": request_id, "ok": True, "data": data}) +async def _ctx(socket: WebSocket, body: dict[str, Any]) -> None: + """A context bundle over the socket, so a client needs no second connection to read.""" + request_id = body.get("id") + name = body.get("context") or "" + fn_names = get_context_groups().get(name) + if not fn_names: + await socket.send_json( + { + "id": request_id, + "ok": False, + "error": {"code": "NOT_FOUND", "message": f"Context '{name}' not found"}, + } + ) + return + + params = body.get("params") or {} + request = _SocketRequest(socket) + try: + bundled = {fn: await execute_function(request, fn, params) for fn in fn_names} + except MizanError as e: + payload: dict[str, Any] = {"code": e.code.value, "message": e.message} + if e.details: + payload["details"] = e.details + await socket.send_json({"id": request_id, "ok": False, "error": payload}) + return + + await socket.send_json({"id": request_id, "ok": True, "data": bundled}) + + _ACTIONS = { "subscribe": _subscribe, "unsubscribe": _unsubscribe, "message": _message, "rpc": _rpc, + "ctx": _ctx, } diff --git a/frontends/mizan-ws-transport/package.json b/frontends/mizan-ws-transport/package.json new file mode 100644 index 0000000..7842c44 --- /dev/null +++ b/frontends/mizan-ws-transport/package.json @@ -0,0 +1,14 @@ +{ + "name": "@mizan/ws-transport", + "version": "0.1.0", + "description": "Mizan transport carrying RPC, context bundles and channel subscriptions over one WebSocket.", + "type": "module", + "main": "src/index.ts", + "exports": { + ".": "./src/index.ts" + }, + "peerDependencies": { + "@mizan/base": "*" + }, + "license": "Elastic-2.0" +} diff --git a/frontends/mizan-ws-transport/src/index.ts b/frontends/mizan-ws-transport/src/index.ts new file mode 100644 index 0000000..f4a19f9 --- /dev/null +++ b/frontends/mizan-ws-transport/src/index.ts @@ -0,0 +1,227 @@ +/** + * `@mizan/ws-transport` — RPC, context bundles and channel subscriptions over one socket. + * + * The same connection carries every direction of traffic, so an app opens one and needs + * no HTTP fallback for reads: + * + * browser -> server + * { action: 'rpc', id, fn, args } + * { action: 'ctx', id, context, params } + * { action: 'subscribe' | 'unsubscribe', channel, params } + * { action: 'message', channel, params, data } + * + * server -> browser + * { id, ok: true, data } // correlated reply + * { id, ok: false, error: { code, message } } + * { channel, params, type, data } // a channel push, uncorrelated + * + * `call` resolves to the `{result, invalidate, merge}` envelope, which is what `mizanCall` + * reads to apply server-driven invalidation — so a mutation sent over the socket + * invalidates exactly as one sent over HTTP. + */ + +import type { MizanTransport } from '@mizan/base' + +type Pending = { resolve: (v: any) => void; reject: (e: any) => void } + +export class MizanSocketError extends Error { + code: string + details?: unknown + + constructor(code: string, message: string, details?: unknown) { + super(message) + this.name = 'MizanSocketError' + this.code = code + this.details = details + } +} + +/** A push from a channel, delivered to every handler subscribed to it. */ +export type ChannelMessage = { + channel: string + params: Record + type: string + data: any +} + +export type ChannelHandler = (msg: ChannelMessage) => void + +export type WsTransportOptions = { + /** Milliseconds between reconnect attempts. 0 disables reconnection. */ + retryMs?: number + /** A reply that never arrives rejects after this many milliseconds. 0 disables. */ + timeoutMs?: number +} + +export class MizanSocket { + private url: string + private socket: WebSocket | null = null + private pending = new Map() + private handlers = new Map>() + // re-sent on every open, so a subscription survives a reconnect + private subscriptions = new Map }>() + private opening: Promise | null = null + private seq = 0 + private retryMs: number + private timeoutMs: number + + constructor(url: string, opts: WsTransportOptions = {}) { + this.url = url + this.retryMs = opts.retryMs ?? 1000 + this.timeoutMs = opts.timeoutMs ?? 30000 + } + + private key(channel: string, params?: Record): string { + const parts = Object.entries(params ?? {}) + .map(([k, v]) => `${k}=${v}`) + .sort() + return parts.length ? `${channel}:${parts.join(':')}` : channel + } + + /** The open socket, connecting if needed. Concurrent callers share one attempt. */ + private connect(): Promise { + if (this.socket && this.socket.readyState === WebSocket.OPEN) return Promise.resolve() + if (this.opening) return this.opening + + this.opening = new Promise((resolve, reject) => { + const socket = new WebSocket(this.url) + this.socket = socket + + socket.onopen = () => { + this.opening = null + for (const sub of this.subscriptions.values()) { + socket.send(JSON.stringify({ action: 'subscribe', ...sub })) + } + resolve() + } + socket.onmessage = (ev) => this.receive(ev.data) + socket.onerror = () => { + this.opening = null + reject(new MizanSocketError('TRANSPORT', `websocket error on ${this.url}`)) + } + socket.onclose = () => { + this.opening = null + this.socket = null + // an in-flight call cannot be answered by a socket that is gone + for (const [, p] of this.pending) { + p.reject(new MizanSocketError('TRANSPORT', 'socket closed before reply')) + } + this.pending.clear() + if (this.retryMs > 0 && this.handlers.size > 0) this.retry() + } + }) + return this.opening + } + + /** Reconnect for the subscriptions still held. A failed attempt is reported and retried. */ + private retry(): void { + setTimeout(() => { + this.connect().catch((e) => { + console.error(`mizan socket: reconnect to ${this.url} failed`, e) + if (this.retryMs > 0 && this.handlers.size > 0) this.retry() + }) + }, this.retryMs) + } + + private receive(raw: string): void { + const msg = JSON.parse(raw) + + if (msg.id !== undefined && this.pending.has(msg.id)) { + const p = this.pending.get(msg.id)! + this.pending.delete(msg.id) + if (msg.ok) p.resolve(msg.data) + else + p.reject( + new MizanSocketError( + msg.error?.code ?? 'INTERNAL_ERROR', + msg.error?.message ?? 'call failed', + msg.error?.details, + ), + ) + return + } + + if (msg.channel !== undefined) { + const handlers = this.handlers.get(this.key(msg.channel, msg.params)) + if (handlers) for (const h of handlers) h(msg as ChannelMessage) + return + } + + // an error with no id belongs to no call — surface it rather than drop it + if (msg.error !== undefined) console.error('mizan socket:', msg.error) + } + + /** Send an action expecting a correlated reply. */ + private async request(action: string, body: Record): Promise { + await this.connect() + const id = String(++this.seq) + return new Promise((resolve, reject) => { + this.pending.set(id, { resolve, reject }) + if (this.timeoutMs > 0) { + setTimeout(() => { + if (this.pending.delete(id)) { + reject(new MizanSocketError('TIMEOUT', `${action} ${id} had no reply`)) + } + }, this.timeoutMs) + } + this.socket!.send(JSON.stringify({ action, id, ...body })) + }) + } + + /** Join a channel. The returned function leaves it. */ + async subscribe( + channel: string, + params: Record | undefined, + handler: ChannelHandler, + ): Promise<() => void> { + const key = this.key(channel, params) + const first = !this.handlers.has(key) + if (first) this.handlers.set(key, new Set()) + this.handlers.get(key)!.add(handler) + + await this.connect() + if (first) { + this.subscriptions.set(key, { channel, params }) + this.socket!.send(JSON.stringify({ action: 'subscribe', channel, params })) + } + + return () => { + const set = this.handlers.get(key) + if (!set) return + set.delete(handler) + if (set.size > 0) return + this.handlers.delete(key) + this.subscriptions.delete(key) + if (this.socket?.readyState === WebSocket.OPEN) { + this.socket.send(JSON.stringify({ action: 'unsubscribe', channel, params })) + } + } + } + + /** Send into a channel. Whether it reaches the group is the channel's decision. */ + async send(channel: string, params: Record | undefined, data: any): Promise { + await this.connect() + this.socket!.send(JSON.stringify({ action: 'message', channel, params, data })) + } + + close(): void { + this.retryMs = 0 + this.socket?.close() + } + + transport(): MizanTransport { + return { + call: (fn, args) => this.request('rpc', { fn, args }), + fetch: (context, params) => this.request('ctx', { context, params }), + } + } +} + +/** The transport, plus the socket it rides, so channels are reachable from the same object. */ +export function wsTransport( + url: string, + opts: WsTransportOptions = {}, +): { transport: MizanTransport; socket: MizanSocket } { + const socket = new MizanSocket(url, opts) + return { transport: socket.transport(), socket } +}