File size: 2,585 Bytes
4e23b01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | /**
* Reverse-RPC handler installer — uniform pattern shared by approval and
* question. The two flows are structurally identical:
*
* 1. WS broadcasts `event.{kind}.requested` with the request payload at
* the top level of `envelope.payload`.
* 2. Test installs `onXxxRequested(handler)`.
* 3. On each request frame, we call `handler(request)` → POST the
* decision/answer to `/sessions/{sid}/{kind}s/{id}`.
*
* Errors from the user handler (or the REST POST) are swallowed into a
* logger.warn — failing reverse-RPC silently means the server will time out
* the approval/question after 60s, which the scenario will observe as a
* timeout in `waitForFrame`. Surfacing those errors here would break the
* "framework auto-responds" contract.
*/
import type { AnyFrame, WsClient } from './ws.js';
export interface ReverseRpcOptions<Req, Res> {
requestEventType: string;
idField: keyof Req & string;
/** REST path under the server API prefix. */
buildPath: (sessionId: string, id: string) => string;
handler: (req: Req) => Promise<Res> | Res;
/** POST helper bound to the right path. */
postResolve: (sessionId: string, id: string, body: Res) => Promise<unknown>;
logger: (level: 'info' | 'warn' | 'error' | 'debug', msg: string, meta?: unknown) => void;
}
/**
* Subscribe to `requestEventType` frames on `ws` and POST the user-supplied
* response. Returns an unsubscribe handle.
*/
export function installReverseRpcHandler<Req, Res>(
ws: WsClient,
opts: ReverseRpcOptions<Req, Res>,
): () => void {
const unsubscribe = ws.onFrame((frame: AnyFrame) => {
if (frame.type !== opts.requestEventType) return;
const payload = frame.payload as Req | undefined;
if (!payload) return;
const sessionId = (payload as { session_id?: string }).session_id;
const id = (payload as Record<string, unknown>)[opts.idField] as string | undefined;
if (!sessionId || !id) {
opts.logger('warn', `reverse-rpc: ${opts.requestEventType} missing session_id/${opts.idField}`, {
payload,
});
return;
}
// Fire-and-forget: the WS handler is sync; we kick off the resolve and
// log async failures.
Promise.resolve()
.then(async () => {
const response = await opts.handler(payload);
await opts.postResolve(sessionId, id, response);
})
.catch((err) => {
opts.logger('warn', `reverse-rpc: ${opts.requestEventType} resolve failed`, {
err: String(err),
sessionId,
id,
});
});
});
return unsubscribe;
}
|