Forward SSE chunks without reserializing events Prevent generic proxy from buffering SSE endpoints Add regression tests for VS Code stream proxy
231 lines
7.9 KiB
TypeScript
231 lines
7.9 KiB
TypeScript
import type { BridgeContext, BridgeResponse } from './bridge';
|
|
import { waitForApiUrl } from './opencode-ready';
|
|
|
|
type BridgeMessageInput = {
|
|
id: string;
|
|
type: string;
|
|
payload?: unknown;
|
|
};
|
|
|
|
type ApiProxyRequestPayload = {
|
|
method?: string;
|
|
path?: string;
|
|
headers?: Record<string, string>;
|
|
bodyBase64?: string;
|
|
};
|
|
|
|
type ApiSessionMessageRequestPayload = {
|
|
path?: string;
|
|
headers?: Record<string, string>;
|
|
bodyText?: string;
|
|
};
|
|
|
|
type ApiProxyResponsePayload = {
|
|
status: number;
|
|
headers: Record<string, string>;
|
|
bodyBase64?: string;
|
|
bodyText?: string;
|
|
};
|
|
|
|
const shouldReturnTextBody = (headers: Headers): boolean => {
|
|
const contentType = headers.get('content-type')?.toLowerCase() || '';
|
|
return contentType.startsWith('application/json')
|
|
|| contentType.startsWith('text/')
|
|
|| contentType.includes('+json');
|
|
};
|
|
|
|
const collectProxyResponseHeaders = (headers: Headers, deps: Pick<ProxyRuntimeDeps, 'collectHeaders'>): Record<string, string> => {
|
|
const result = deps.collectHeaders(headers);
|
|
delete result['content-length'];
|
|
delete result['content-encoding'];
|
|
delete result['transfer-encoding'];
|
|
return result;
|
|
};
|
|
|
|
const isSseProxyPath = (requestPath: string): boolean => {
|
|
try {
|
|
const parsed = new URL(requestPath, 'https://openchamber.invalid');
|
|
return parsed.pathname === '/event' || parsed.pathname === '/global/event';
|
|
} catch {
|
|
return requestPath === '/event' || requestPath === '/global/event';
|
|
}
|
|
};
|
|
|
|
type ProxyRuntimeDeps = {
|
|
tryHandleLocalFsProxy: (method: string, requestPath: string) => Promise<ApiProxyResponsePayload | null>;
|
|
buildUnavailableApiResponse: () => ApiProxyResponsePayload;
|
|
sanitizeForwardHeaders: (input: Record<string, string> | undefined) => Record<string, string>;
|
|
collectHeaders: (headers: Headers) => Record<string, string>;
|
|
base64EncodeUtf8: (text: string) => string;
|
|
};
|
|
|
|
export async function handleProxyBridgeMessage(
|
|
message: BridgeMessageInput,
|
|
ctx: BridgeContext | undefined,
|
|
deps: ProxyRuntimeDeps,
|
|
): Promise<BridgeResponse | null> {
|
|
const { id, type, payload } = message;
|
|
|
|
switch (type) {
|
|
case 'api:proxy': {
|
|
const { method, path: requestPath, headers, bodyBase64 } = (payload || {}) as ApiProxyRequestPayload;
|
|
const normalizedMethod = typeof method === 'string' && method.trim() ? method.trim().toUpperCase() : 'GET';
|
|
const normalizedPath =
|
|
typeof requestPath === 'string' && requestPath.trim().length > 0
|
|
? requestPath.trim().startsWith('/')
|
|
? requestPath.trim()
|
|
: `/${requestPath.trim()}`
|
|
: '/';
|
|
|
|
if (isSseProxyPath(normalizedPath)) {
|
|
const data: ApiProxyResponsePayload = {
|
|
status: 400,
|
|
headers: { 'content-type': 'application/json' },
|
|
bodyText: JSON.stringify({ error: 'SSE requests must use api:sse:start' }),
|
|
};
|
|
return { id, type, success: true, data };
|
|
}
|
|
|
|
const localFsResponse = await deps.tryHandleLocalFsProxy(normalizedMethod, normalizedPath);
|
|
if (localFsResponse) {
|
|
return { id, type, success: true, data: localFsResponse };
|
|
}
|
|
|
|
const apiUrl = await waitForApiUrl(ctx?.manager);
|
|
if (!apiUrl) {
|
|
const data = deps.buildUnavailableApiResponse();
|
|
return { id, type, success: true, data };
|
|
}
|
|
|
|
const base = `${apiUrl.replace(/\/+$/, '')}/`;
|
|
const targetUrl = new URL(normalizedPath.replace(/^\/+/, ''), base).toString();
|
|
const requestHeaders: Record<string, string> = {
|
|
...deps.sanitizeForwardHeaders(headers),
|
|
...ctx?.manager?.getOpenCodeAuthHeaders(),
|
|
};
|
|
|
|
try {
|
|
const response = await fetch(targetUrl, {
|
|
method: normalizedMethod,
|
|
headers: requestHeaders,
|
|
body:
|
|
typeof bodyBase64 === 'string' && bodyBase64.length > 0 && normalizedMethod !== 'GET' && normalizedMethod !== 'HEAD'
|
|
? Buffer.from(bodyBase64, 'base64')
|
|
: undefined,
|
|
});
|
|
|
|
const responseHeaders = collectProxyResponseHeaders(response.headers, deps);
|
|
if (shouldReturnTextBody(response.headers)) {
|
|
const bodyText = await response.text();
|
|
const data: ApiProxyResponsePayload = {
|
|
status: response.status,
|
|
headers: responseHeaders,
|
|
bodyText,
|
|
};
|
|
|
|
return { id, type, success: true, data };
|
|
}
|
|
|
|
const arrayBuffer = await response.arrayBuffer();
|
|
const data: ApiProxyResponsePayload = {
|
|
status: response.status,
|
|
headers: responseHeaders,
|
|
bodyBase64: Buffer.from(arrayBuffer).toString('base64'),
|
|
};
|
|
|
|
return { id, type, success: true, data };
|
|
} catch (error) {
|
|
const body = JSON.stringify({
|
|
error: error instanceof Error ? error.message : 'Failed to reach OpenCode API',
|
|
});
|
|
const data: ApiProxyResponsePayload = {
|
|
status: 502,
|
|
headers: { 'content-type': 'application/json' },
|
|
bodyText: body,
|
|
};
|
|
return { id, type, success: true, data };
|
|
}
|
|
}
|
|
|
|
case 'api:session:message': {
|
|
const apiUrl = await waitForApiUrl(ctx?.manager);
|
|
if (!apiUrl) {
|
|
const data = deps.buildUnavailableApiResponse();
|
|
return { id, type, success: true, data };
|
|
}
|
|
|
|
const { path: requestPath, headers, bodyText } = (payload || {}) as ApiSessionMessageRequestPayload;
|
|
const normalizedPath =
|
|
typeof requestPath === 'string' && requestPath.trim().length > 0
|
|
? requestPath.trim().startsWith('/')
|
|
? requestPath.trim()
|
|
: `/${requestPath.trim()}`
|
|
: '/';
|
|
|
|
if (!/^\/session\/[^/]+\/message(?:\?.*)?$/.test(normalizedPath)) {
|
|
const body = JSON.stringify({ error: 'Invalid session message proxy path' });
|
|
const data: ApiProxyResponsePayload = {
|
|
status: 400,
|
|
headers: { 'content-type': 'application/json' },
|
|
bodyBase64: deps.base64EncodeUtf8(body),
|
|
};
|
|
return { id, type, success: true, data };
|
|
}
|
|
|
|
const base = `${apiUrl.replace(/\/+$/, '')}/`;
|
|
const targetUrl = new URL(normalizedPath.replace(/^\/+/, ''), base).toString();
|
|
const requestHeaders: Record<string, string> = {
|
|
...deps.sanitizeForwardHeaders(headers),
|
|
...ctx?.manager?.getOpenCodeAuthHeaders(),
|
|
};
|
|
|
|
try {
|
|
const response = await fetch(targetUrl, {
|
|
method: 'POST',
|
|
headers: requestHeaders,
|
|
body: typeof bodyText === 'string' ? bodyText : '',
|
|
signal: AbortSignal.timeout(45000),
|
|
});
|
|
|
|
const responseHeaders = collectProxyResponseHeaders(response.headers, deps);
|
|
if (shouldReturnTextBody(response.headers)) {
|
|
const bodyText = await response.text();
|
|
const data: ApiProxyResponsePayload = {
|
|
status: response.status,
|
|
headers: responseHeaders,
|
|
bodyText,
|
|
};
|
|
|
|
return { id, type, success: true, data };
|
|
}
|
|
|
|
const arrayBuffer = await response.arrayBuffer();
|
|
const data: ApiProxyResponsePayload = {
|
|
status: response.status,
|
|
headers: responseHeaders,
|
|
bodyBase64: Buffer.from(arrayBuffer).toString('base64'),
|
|
};
|
|
|
|
return { id, type, success: true, data };
|
|
} catch (error) {
|
|
const isTimeout =
|
|
error instanceof Error &&
|
|
((error as Error & { name?: string }).name === 'TimeoutError' ||
|
|
(error as Error & { name?: string }).name === 'AbortError');
|
|
const body = JSON.stringify({
|
|
error: isTimeout ? 'OpenCode message forward timed out' : error instanceof Error ? error.message : 'OpenCode message forward failed',
|
|
});
|
|
const data: ApiProxyResponsePayload = {
|
|
status: isTimeout ? 504 : 503,
|
|
headers: { 'content-type': 'application/json' },
|
|
bodyText: body,
|
|
};
|
|
return { id, type, success: true, data };
|
|
}
|
|
}
|
|
|
|
default:
|
|
return null;
|
|
}
|
|
}
|