diff --git a/src/node/orpc/router.ts b/src/node/orpc/router.ts index 4424442aeb..9f70bc2f96 100644 --- a/src/node/orpc/router.ts +++ b/src/node/orpc/router.ts @@ -1,3 +1,22 @@ +/** + * oRPC router: procedure definitions only — behavior lives in services. + * + * Handler convention (Effect migration, phases 1-4): new unary procedures + * should ride `handlerGen` from `@orpc/experimental-effect` and yield a + * wire-shaped Effect exposed by the backing service (see `memory.*`, + * `providers.*` mutations, and `muxGateway*` for the house pattern). Plain + * async handlers remain appropriate only when: + * - the backing service is still Promise-based (convert the service surface + * first; do not wrap Promises in Effect at the router), + * - the procedure returns an event iterator (subscriptions — handlerGen + * cannot produce those until an Effect Stream bridge exists), or + * - the handler is a trivial synchronous read. + * + * handlerGen makes handlers interruptible on client abort: before converting + * a mutation, audit its abort-atomicity and keep multi-step writes + * uninterruptible in the service pipeline (see asAtomicMutation in + * providerService.ts and startDesktopFlowEffect in muxGatewayOauthService.ts). + */ import { os } from "@orpc/server"; import * as schemas from "@/common/orpc/schemas"; import type { ORPCContext } from "./context"; @@ -549,14 +568,26 @@ export const router = (authToken?: string) => { getAccountStatus: t .input(schemas.muxGateway.getAccountStatus.input) .output(schemas.muxGateway.getAccountStatus.output) - .handler(({ context }) => context.muxGatewayOauthService.getAccountStatus()), + .handler( + handlerGen(function* ({ context }) { + return yield* context.muxGatewayOauthService.getAccountStatusEffect(); + }) + ), }, muxGatewayOauth: { + // startDesktopFlow rides handlerGen; its service pipeline is + // uninterruptible (see startDesktopFlowEffect) so a client abort cannot + // leak the loopback server. waitFor/cancel stay plain handlers until the + // promise-native OAuthFlowManager grows an Effect surface. startDesktopFlow: t .input(schemas.muxGatewayOauth.startDesktopFlow.input) .output(schemas.muxGatewayOauth.startDesktopFlow.output) - .handler(({ context }) => context.muxGatewayOauthService.startDesktopFlow()), + .handler( + handlerGen(function* ({ context }) { + return yield* context.muxGatewayOauthService.startDesktopFlowEffect(); + }) + ), waitForDesktopFlow: t .input(schemas.muxGatewayOauth.waitForDesktopFlow.input) .output(schemas.muxGatewayOauth.waitForDesktopFlow.output) diff --git a/src/node/services/muxGatewayOauthService.ts b/src/node/services/muxGatewayOauthService.ts index deb481c01a..8aa6813497 100644 --- a/src/node/services/muxGatewayOauthService.ts +++ b/src/node/services/muxGatewayOauthService.ts @@ -94,26 +94,39 @@ export class MuxGatewayOauthService { }, string > + > { + return Effect.runPromise(this.getAccountStatusEffect()); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. Left + * interruptible: the balance fetch is a pure read, and the session-expired + * credential clear is a single best-effort promise that runs to completion + * even if the fiber is interrupted while awaiting it. + */ + getAccountStatusEffect(): Effect.Effect< + Result< + { remaining_microdollars: number; ai_gateway_concurrent_requests_per_user: number }, + string + > > { // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` const self = this; - return Effect.runPromise( - toWireResult( - this.getAccountStatusEffect().pipe( - Effect.catchTag("MuxGatewaySessionExpiredError", () => - Effect.gen(function* () { - yield* self.clearStoredCredentials(); - return yield* Effect.fail( - new MuxGatewayOAuthError({ reason: MUX_GATEWAY_SESSION_EXPIRED_MESSAGE }) - ); - }) - ) + return toWireResult( + this.fetchAccountStatusEffect().pipe( + Effect.catchTag("MuxGatewaySessionExpiredError", () => + Effect.gen(function* () { + yield* self.clearStoredCredentials(); + return yield* Effect.fail( + new MuxGatewayOAuthError({ reason: MUX_GATEWAY_SESSION_EXPIRED_MESSAGE }) + ); + }) ) ) ); } - private getAccountStatusEffect(): Effect.Effect< + private fetchAccountStatusEffect(): Effect.Effect< { remaining_microdollars: number; ai_gateway_concurrent_requests_per_user: number }, MuxGatewaySessionExpiredError | MuxGatewayOAuthError > { @@ -225,10 +238,24 @@ export class MuxGatewayOauthService { async startDesktopFlow(): Promise< Result<{ flowId: string; authorizeUrl: string; redirectUri: string }, string> > { - return Effect.runPromise(toWireResult(this.startDesktopFlowEffect())); + return Effect.runPromise(this.startDesktopFlowEffect()); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. Uninterruptible + * (mirrors asAtomicMutation in providerService.ts): a client abort between + * the loopback-server acquisition and `desktopFlows.register` would leak the + * server with nothing left to close it. Flow startup is quick and local, so + * running it to completion on abort is cheap; an abandoned flow still + * self-cleans via the registered timeout. + */ + startDesktopFlowEffect(): Effect.Effect< + Result<{ flowId: string; authorizeUrl: string; redirectUri: string }, string> + > { + return Effect.uninterruptible(toWireResult(this.launchDesktopFlowEffect())); } - private startDesktopFlowEffect(): Effect.Effect< + private launchDesktopFlowEffect(): Effect.Effect< { flowId: string; authorizeUrl: string; redirectUri: string }, MuxGatewayOAuthError > {