feat(frontend): adapter web HTTP+WebSocket derrière les ports (#13)
Lot F1 du chantier server/client mode : nouvel adaptateur web branché derrière les ports d'invocation et de flux live, permettant au frontend de dialoguer avec le backend via HTTP + WebSocket en mode client/serveur. Le mode desktop (Tauri IPC) reste inchangé. - frontend/src/adapters/http : invoker HTTP, client live WebSocket, gateways request/response et stream, frames, garde unsupported (7 fichiers + 2 tests). - frontend/src/app : câblage DI (di.tsx) et son test, typage vite-env.d.ts. Validé : build vert, garde no-direct-invoke verte, 724 tests verts, desktop inchangé. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
232
frontend/src/adapters/http/wsLiveClient.ts
Normal file
232
frontend/src/adapters/http/wsLiveClient.ts
Normal file
@ -0,0 +1,232 @@
|
||||
/**
|
||||
* WebSocket live client skeleton for the web transport — ticket #13, lot F1.
|
||||
*
|
||||
* Owns a single WS connection to `{wsUrl}/ws/live` and multiplexes over it, per
|
||||
* the B0 draft: PTY output, structured chat output and the low-frequency domain
|
||||
* event stream (`event.domain`, replacing Tauri `listen("domain://event")`).
|
||||
*
|
||||
* **F1 scope = skeleton.** The connection, frame envelope, request/reply
|
||||
* correlation by `id`, and the per-session output routing are implemented and
|
||||
* unit-testable (inject a fake WebSocket factory). What is deliberately deferred
|
||||
* to **F3/B5** (when a server exists — B3/B4): reconnection/backpressure, exact
|
||||
* `open`/`launch` reply shape, sequence-gap replay policy, and the full xterm
|
||||
* round-trip. Those are marked `TODO(F3/B5)`.
|
||||
*
|
||||
* Lives in `src/adapters/**`; touches no `@tauri-apps/api`.
|
||||
*/
|
||||
|
||||
import type { DomainEvent } from "@/domain";
|
||||
import {
|
||||
base64ToBytes,
|
||||
bytesToBase64,
|
||||
type ClientFrame,
|
||||
type ClientFrameKind,
|
||||
type OutputPayload,
|
||||
type ServerFrame,
|
||||
} from "./frames";
|
||||
|
||||
/** Minimal WebSocket surface used here; injectable so tests pass a fake. */
|
||||
export interface WebSocketLike {
|
||||
send(data: string): void;
|
||||
close(): void;
|
||||
onopen: (() => void) | null;
|
||||
onmessage: ((event: { data: string }) => void) | null;
|
||||
onerror: ((event: unknown) => void) | null;
|
||||
onclose: (() => void) | null;
|
||||
}
|
||||
|
||||
/** Factory building a {@link WebSocketLike} for a URL (defaults to global WS). */
|
||||
export type WebSocketFactory = (url: string) => WebSocketLike;
|
||||
|
||||
/** Configuration for the live client. */
|
||||
export interface WsLiveClientConfig {
|
||||
/** Base WS URL, e.g. `wss://host:port`. No trailing slash. */
|
||||
wsUrl: string;
|
||||
/** Bearer token (ticket #13 auth). B0 note: prefer header/cookie over URL. */
|
||||
token?: string;
|
||||
/** Injected WebSocket factory (defaults to `new WebSocket(url)`). */
|
||||
socketFactory?: WebSocketFactory;
|
||||
}
|
||||
|
||||
let frameCounter = 0;
|
||||
/** Monotonic client frame id (unique per client session). */
|
||||
function nextFrameId(): string {
|
||||
frameCounter += 1;
|
||||
return `c${frameCounter}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Manages the single live WebSocket. Callers subscribe an output sink per PTY
|
||||
* session and a single domain-event handler; the client routes inbound frames.
|
||||
*/
|
||||
export class WsLiveClient {
|
||||
private readonly wsUrl: string;
|
||||
private readonly token?: string;
|
||||
private readonly socketFactory: WebSocketFactory;
|
||||
|
||||
private socket: WebSocketLike | null = null;
|
||||
private opening: Promise<void> | null = null;
|
||||
|
||||
/** Per-session PTY output sinks (sessionId → onData). */
|
||||
private readonly outputSinks = new Map<string, (bytes: Uint8Array) => void>();
|
||||
/** Pending command acknowledgements, keyed by client frame id. */
|
||||
private readonly pending = new Map<
|
||||
string,
|
||||
{ resolve: (frame: ServerFrame) => void; reject: (err: unknown) => void }
|
||||
>();
|
||||
/** The single domain-event handler (set by the system gateway). */
|
||||
private domainEventHandler: ((event: DomainEvent) => void) | null = null;
|
||||
|
||||
constructor(config: WsLiveClientConfig) {
|
||||
this.wsUrl = config.wsUrl.replace(/\/+$/, "");
|
||||
this.token = config.token;
|
||||
this.socketFactory =
|
||||
config.socketFactory ??
|
||||
((url: string) => new WebSocket(url) as unknown as WebSocketLike);
|
||||
}
|
||||
|
||||
/** Registers the domain-event handler (replaces any previous one). */
|
||||
setDomainEventHandler(handler: (event: DomainEvent) => void): void {
|
||||
this.domainEventHandler = handler;
|
||||
}
|
||||
|
||||
/** Clears the domain-event handler. */
|
||||
clearDomainEventHandler(): void {
|
||||
this.domainEventHandler = null;
|
||||
}
|
||||
|
||||
/** Registers a per-session PTY output sink. */
|
||||
setOutputSink(sessionId: string, onData: (bytes: Uint8Array) => void): void {
|
||||
this.outputSinks.set(sessionId, onData);
|
||||
}
|
||||
|
||||
/** Drops a per-session PTY output sink (view detached). */
|
||||
removeOutputSink(sessionId: string): void {
|
||||
this.outputSinks.delete(sessionId);
|
||||
}
|
||||
|
||||
/** Ensures the socket is connected, connecting on first use. */
|
||||
async ensureConnected(): Promise<void> {
|
||||
if (this.socket) return;
|
||||
if (this.opening) return this.opening;
|
||||
|
||||
this.opening = new Promise<void>((resolve, reject) => {
|
||||
// B0 auth note: token should ride an `Authorization` header / secure
|
||||
// cookie at upgrade, NOT a URL secret. Browsers can't set WS upgrade
|
||||
// headers, so the token placement is a contract point to confirm (see the
|
||||
// F1 report). The query below is only a placeholder for the skeleton.
|
||||
const url = this.token
|
||||
? `${this.wsUrl}/ws/live?token=${encodeURIComponent(this.token)}`
|
||||
: `${this.wsUrl}/ws/live`;
|
||||
const socket = this.socketFactory(url);
|
||||
socket.onopen = () => {
|
||||
this.socket = socket;
|
||||
resolve();
|
||||
};
|
||||
socket.onerror = (event) => {
|
||||
this.opening = null;
|
||||
reject(event);
|
||||
};
|
||||
socket.onclose = () => {
|
||||
// TODO(F3/B5): reconnection + re-attach of live sessions. For F1 we drop
|
||||
// the socket so a later call reconnects fresh.
|
||||
this.socket = null;
|
||||
this.rejectAllPending({ code: "WS_CLOSED", message: "socket closed" });
|
||||
};
|
||||
socket.onmessage = (event) => this.handleMessage(event.data);
|
||||
});
|
||||
return this.opening;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a client frame and resolves with its acknowledgement frame (the server
|
||||
* frame whose `replyTo` equals this frame's `id`). Fire-and-forget frames
|
||||
* (input/resize/detach) can ignore the returned promise.
|
||||
*/
|
||||
async send(
|
||||
kind: ClientFrameKind,
|
||||
payload: Record<string, unknown>,
|
||||
): Promise<ServerFrame> {
|
||||
await this.ensureConnected();
|
||||
const socket = this.socket;
|
||||
if (!socket) {
|
||||
const err = { code: "WS_CLOSED", message: "socket not connected" };
|
||||
throw err;
|
||||
}
|
||||
const id = nextFrameId();
|
||||
const frame: ClientFrame = { id, kind, payload };
|
||||
const ack = new Promise<ServerFrame>((resolve, reject) => {
|
||||
this.pending.set(id, { resolve, reject });
|
||||
});
|
||||
socket.send(JSON.stringify(frame));
|
||||
return ack;
|
||||
}
|
||||
|
||||
/** Sends a fire-and-forget frame (no acknowledgement awaited). */
|
||||
async sendFireAndForget(
|
||||
kind: ClientFrameKind,
|
||||
payload: Record<string, unknown>,
|
||||
): Promise<void> {
|
||||
await this.ensureConnected();
|
||||
const id = nextFrameId();
|
||||
const frame: ClientFrame = { id, kind, payload };
|
||||
this.socket?.send(JSON.stringify(frame));
|
||||
}
|
||||
|
||||
/** Convenience: base64-encode bytes for an `input` frame. */
|
||||
static encodeInput(bytes: Uint8Array): string {
|
||||
return bytesToBase64(bytes);
|
||||
}
|
||||
|
||||
/** Closes the socket and rejects everything pending. */
|
||||
dispose(): void {
|
||||
this.rejectAllPending({ code: "WS_CLOSED", message: "client disposed" });
|
||||
this.outputSinks.clear();
|
||||
this.domainEventHandler = null;
|
||||
this.socket?.close();
|
||||
this.socket = null;
|
||||
this.opening = null;
|
||||
}
|
||||
|
||||
private rejectAllPending(err: unknown): void {
|
||||
for (const { reject } of this.pending.values()) reject(err);
|
||||
this.pending.clear();
|
||||
}
|
||||
|
||||
private handleMessage(data: string): void {
|
||||
let frame: ServerFrame;
|
||||
try {
|
||||
frame = JSON.parse(data) as ServerFrame;
|
||||
} catch {
|
||||
return; // Ignore malformed frames in the skeleton.
|
||||
}
|
||||
|
||||
// Route unsolicited streams first (no replyTo).
|
||||
switch (frame.kind) {
|
||||
case "terminal.output": {
|
||||
const payload = frame.payload as unknown as OutputPayload;
|
||||
const sink = this.outputSinks.get(payload.sessionId);
|
||||
// TODO(F3/B5): honour `seq` ordering + gap detection for precise replay.
|
||||
if (sink) sink(base64ToBytes(payload.bytesBase64));
|
||||
return;
|
||||
}
|
||||
case "event.domain": {
|
||||
// `payload` is a DomainEventDto (kind-tagged); forward as-is.
|
||||
this.domainEventHandler?.(frame.payload as unknown as DomainEvent);
|
||||
return;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
|
||||
// Otherwise it is a reply/ack correlated by `replyTo`.
|
||||
if (frame.replyTo) {
|
||||
const waiter = this.pending.get(frame.replyTo);
|
||||
if (waiter) {
|
||||
this.pending.delete(frame.replyTo);
|
||||
if (frame.kind === "error") waiter.reject(frame.payload);
|
||||
else waiter.resolve(frame);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user