refactor(server): unify instance httpapi middleware routing
Unify declared instance HTTP API endpoints under typed middleware routing, including event streaming and PTY WebSocket connect handling.\n\nPreserve PTY connect compatibility by checking missing PTYs before parsing optional cursor and ticket query fields, with regression coverage.
This commit is contained in:
@@ -10,13 +10,13 @@ import {
|
||||
PTY_CONNECT_TOKEN_HEADER,
|
||||
PTY_CONNECT_TOKEN_HEADER_VALUE,
|
||||
} from "@/server/shared/pty-ticket"
|
||||
import { Effect } from "effect"
|
||||
import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
|
||||
import { Effect, Option, Schema } from "effect"
|
||||
import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
|
||||
import { HttpApiBuilder } from "effect/unstable/httpapi"
|
||||
import * as Socket from "effect/unstable/socket/Socket"
|
||||
import { InstanceHttpApi } from "../api"
|
||||
import * as ApiError from "../errors"
|
||||
import { CursorQuery, Params, PtyPaths } from "../groups/pty"
|
||||
import { CursorQuery, PtyConnectApi } from "../groups/pty"
|
||||
import { WebSocketTracker } from "../websocket-tracker"
|
||||
|
||||
function validOrigin(request: HttpServerRequest.HttpServerRequest, opts: CorsOptions | undefined) {
|
||||
@@ -121,37 +121,39 @@ export const ptyHandlers = HttpApiBuilder.group(InstanceHttpApi, "pty", (handler
|
||||
}),
|
||||
)
|
||||
|
||||
export const ptyConnectRoute = HttpRouter.use((router) =>
|
||||
export const ptyConnectHandlers = HttpApiBuilder.group(PtyConnectApi, "pty-connect", (handlers) =>
|
||||
Effect.gen(function* () {
|
||||
const pty = yield* Pty.Service
|
||||
const tickets = yield* PtyTicket.Service
|
||||
const cors = yield* CorsConfig
|
||||
yield* router.add(
|
||||
"GET",
|
||||
PtyPaths.connect,
|
||||
Effect.gen(function* () {
|
||||
const params = yield* HttpRouter.schemaPathParams(Params)
|
||||
const exists = yield* pty.get(params.ptyID).pipe(
|
||||
|
||||
return handlers.handleRaw(
|
||||
"connect",
|
||||
Effect.fn("PtyHttpApi.connect")(function* (ctx: {
|
||||
params: { ptyID: PtyID }
|
||||
request: HttpServerRequest.HttpServerRequest
|
||||
}) {
|
||||
const exists = yield* pty.get(ctx.params.ptyID).pipe(
|
||||
Effect.as(true),
|
||||
Effect.catchTag("Pty.NotFoundError", () => Effect.succeed(false)),
|
||||
)
|
||||
if (!exists) return HttpServerResponse.empty({ status: 404 })
|
||||
|
||||
const query = yield* HttpServerRequest.schemaSearchParams(CursorQuery)
|
||||
const request = yield* HttpServerRequest.HttpServerRequest
|
||||
const ticket = new URL(request.url, "http://localhost").searchParams.get(PTY_CONNECT_TICKET_QUERY)
|
||||
const query = Schema.decodeUnknownOption(CursorQuery)(yield* HttpServerRequest.ParsedSearchParams)
|
||||
if (Option.isNone(query)) return HttpServerResponse.empty({ status: 400 })
|
||||
const ticket = new URL(ctx.request.url, "http://localhost").searchParams.get(PTY_CONNECT_TICKET_QUERY)
|
||||
if (ticket) {
|
||||
const valid = validOrigin(request, cors)
|
||||
? yield* tickets.consume({ ticket, ptyID: params.ptyID, ...(yield* PtyTicket.scope) })
|
||||
const valid = validOrigin(ctx.request, cors)
|
||||
? yield* tickets.consume({ ticket, ptyID: ctx.params.ptyID, ...(yield* PtyTicket.scope) })
|
||||
: false
|
||||
if (!valid) return HttpServerResponse.empty({ status: 403 })
|
||||
}
|
||||
const parsedCursor = query.cursor === undefined ? undefined : Number(query.cursor)
|
||||
const parsedCursor = query.value.cursor === undefined ? undefined : Number(query.value.cursor)
|
||||
const cursor =
|
||||
parsedCursor !== undefined && Number.isSafeInteger(parsedCursor) && parsedCursor >= -1
|
||||
? parsedCursor
|
||||
: undefined
|
||||
const socket = yield* Effect.orDie(request.upgrade)
|
||||
const socket = yield* Effect.orDie(ctx.request.upgrade)
|
||||
const write = yield* socket.writer
|
||||
const closeAccepted = (event: Socket.CloseEvent) =>
|
||||
socket
|
||||
@@ -186,7 +188,7 @@ export const ptyConnectRoute = HttpRouter.use((router) =>
|
||||
},
|
||||
}
|
||||
const handler = yield* pty
|
||||
.connect(params.ptyID, adapter, cursor)
|
||||
.connect(ctx.params.ptyID, adapter, cursor)
|
||||
.pipe(
|
||||
Effect.catchTag("Pty.NotFoundError", () =>
|
||||
closeAccepted(new Socket.CloseEvent(4404, "session not found")).pipe(Effect.as(undefined)),
|
||||
@@ -194,12 +196,8 @@ export const ptyConnectRoute = HttpRouter.use((router) =>
|
||||
)
|
||||
if (!handler) return HttpServerResponse.empty()
|
||||
|
||||
// No `pending[]`-style early-frame buffer (the legacy handler had one).
|
||||
// `request.upgrade` returns a Socket without running the WS handshake; the
|
||||
// handshake fires inside `socket.runRaw` below, AFTER `pty.connect` resolves
|
||||
// and the message callback is registered. The client therefore can't fire
|
||||
// `open` and start sending until the listener is already wired. Don't move
|
||||
// `runRaw` ahead of `pty.connect` without re-introducing a buffer.
|
||||
// The handshake runs inside `socket.runRaw`, after the input callback is
|
||||
// registered, so the client cannot send frames before PTY input is wired.
|
||||
yield* socket
|
||||
.runRaw((message) => handlePtyInput(handler, message))
|
||||
.pipe(
|
||||
|
||||
Reference in New Issue
Block a user