feat(FN-1806): add node proxy API routes for remote node data fetching
- Add proxy API routes for cross-node data fetching (tasks, agents, stats, health) - Implement secure proxy endpoint with target validation and error handling - Add comprehensive test suite covering proxy route functionality - Update memory documentation with cross-node proxy implementation
This commit is contained in:
Fusion
574
packages/dashboard/src/__tests__/routes-proxy.test.ts
Normal file
574
packages/dashboard/src/__tests__/routes-proxy.test.ts
Normal file
@@ -0,0 +1,574 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { Task } from "@fusion/core";
|
||||
import { request } from "../test-request.js";
|
||||
import { createServer } from "../server.js";
|
||||
|
||||
const mockInit = vi.fn().mockResolvedValue(undefined);
|
||||
const mockClose = vi.fn().mockResolvedValue(undefined);
|
||||
const mockGetNode = vi.fn();
|
||||
|
||||
vi.mock("@fusion/core", async () => {
|
||||
const actual = await vi.importActual<typeof import("@fusion/core")>("@fusion/core");
|
||||
return {
|
||||
...actual,
|
||||
CentralCore: vi.fn().mockImplementation(() => ({
|
||||
init: mockInit,
|
||||
close: mockClose,
|
||||
getNode: mockGetNode,
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
class MockStore extends EventEmitter {
|
||||
getRootDir(): string {
|
||||
return "/tmp/fn-1806";
|
||||
}
|
||||
|
||||
getFusionDir(): string {
|
||||
return "/tmp/fn-1806/.fusion";
|
||||
}
|
||||
|
||||
getDatabase() {
|
||||
return {
|
||||
exec: vi.fn(),
|
||||
prepare: vi.fn().mockReturnValue({ run: vi.fn().mockReturnValue({ changes: 0 }), get: vi.fn(), all: vi.fn().mockReturnValue([]) }),
|
||||
};
|
||||
}
|
||||
|
||||
getMissionStore() {
|
||||
return {
|
||||
listMissions: vi.fn().mockResolvedValue([]),
|
||||
createMission: vi.fn(),
|
||||
getMission: vi.fn(),
|
||||
updateMission: vi.fn(),
|
||||
deleteMission: vi.fn(),
|
||||
listTemplates: vi.fn().mockResolvedValue([]),
|
||||
createTemplate: vi.fn(),
|
||||
getTemplate: vi.fn(),
|
||||
updateTemplate: vi.fn(),
|
||||
deleteTemplate: vi.fn(),
|
||||
instantiateMission: vi.fn(),
|
||||
};
|
||||
}
|
||||
|
||||
async listTasks(): Promise<Task[]> {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function makeNode(overrides: Partial<Record<string, unknown>> = {}) {
|
||||
return {
|
||||
id: "node_local",
|
||||
name: "local-node",
|
||||
type: "local",
|
||||
status: "online",
|
||||
maxConcurrent: 2,
|
||||
capabilities: ["executor"],
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
describe("Node proxy routes", () => {
|
||||
const app = createServer(new MockStore() as any);
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
mockGetNode.mockResolvedValue(undefined);
|
||||
vi.mock("@fusion/core", async () => {
|
||||
const actual = await vi.importActual<typeof import("@fusion/core")>("@fusion/core");
|
||||
return {
|
||||
...actual,
|
||||
CentralCore: vi.fn().mockImplementation(() => ({
|
||||
init: mockInit,
|
||||
close: mockClose,
|
||||
getNode: mockGetNode,
|
||||
})),
|
||||
};
|
||||
});
|
||||
});
|
||||
|
||||
// ── Helper to create a mock fetch response ─────────────────────────────
|
||||
|
||||
function makeMockFetchResponse(options: {
|
||||
status?: number;
|
||||
ok?: boolean;
|
||||
body?: unknown;
|
||||
headers?: Record<string, string>;
|
||||
streamChunks?: string[];
|
||||
}): Promise<Response> {
|
||||
const {
|
||||
status = 200,
|
||||
ok = true,
|
||||
body,
|
||||
headers = { "content-type": "application/json" },
|
||||
streamChunks = [],
|
||||
} = options;
|
||||
|
||||
// Build chunks: if body is provided, encode it as JSON; otherwise use streamChunks
|
||||
const chunks =
|
||||
body !== undefined && streamChunks.length === 0
|
||||
? [JSON.stringify(body)]
|
||||
: streamChunks;
|
||||
|
||||
const readable = new ReadableStream({
|
||||
start(controller) {
|
||||
for (const chunk of chunks) {
|
||||
controller.enqueue(new TextEncoder().encode(chunk));
|
||||
}
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
const mockResponse = {
|
||||
ok,
|
||||
status,
|
||||
statusText: status === 200 ? "OK" : "Error",
|
||||
headers: new Map(Object.entries(headers)),
|
||||
body: readable,
|
||||
} as unknown as Response;
|
||||
|
||||
return Promise.resolve(mockResponse);
|
||||
}
|
||||
|
||||
// ── Successful proxy tests ─────────────────────────────────────────────
|
||||
|
||||
describe("GET /api/proxy/:nodeId/health", () => {
|
||||
it("returns forwarded JSON from remote node health endpoint", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
makeMockFetchResponse({
|
||||
status: 200,
|
||||
ok: true,
|
||||
body: { status: "ok", version: "1.0.0" },
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_remote/health");
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body).toEqual({ status: "ok", version: "1.0.0" });
|
||||
expect(mockInit).toHaveBeenCalled();
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("GET /api/proxy/:nodeId/projects", () => {
|
||||
it("returns forwarded JSON from remote node projects endpoint", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
makeMockFetchResponse({
|
||||
status: 200,
|
||||
ok: true,
|
||||
body: [
|
||||
{ id: "proj_1", name: "Project 1", path: "/tmp/p1" },
|
||||
{ id: "proj_2", name: "Project 2", path: "/tmp/p2" },
|
||||
],
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_remote/projects");
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect((res.body as unknown[])).toHaveLength(2);
|
||||
expect((res.body as unknown[])[0]).toEqual({ id: "proj_1", name: "Project 1", path: "/tmp/p1" });
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("GET /api/proxy/:nodeId/tasks", () => {
|
||||
it("forwards query params to remote node tasks endpoint", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
makeMockFetchResponse({
|
||||
status: 200,
|
||||
ok: true,
|
||||
body: [],
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(
|
||||
app,
|
||||
"GET",
|
||||
"/api/proxy/node_remote/tasks?projectId=proj_123&q=test+query",
|
||||
);
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
|
||||
// Verify fetch was called with the correct URL containing forwarded query params
|
||||
const fetchCall = vi.mocked(fetch).mock.calls[0];
|
||||
expect(fetchCall[0]).toBe(
|
||||
"http://remote:4040/api/tasks?projectId=proj_123&q=test+query",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("GET /api/proxy/:nodeId/project-health", () => {
|
||||
it("forwards projectId query param to remote node project-health endpoint", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
makeMockFetchResponse({
|
||||
status: 200,
|
||||
ok: true,
|
||||
body: { healthy: true, tasksDone: 42, tasksTotal: 100 },
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(
|
||||
app,
|
||||
"GET",
|
||||
"/api/proxy/node_remote/project-health?projectId=proj_456",
|
||||
);
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body).toEqual({ healthy: true, tasksDone: 42, tasksTotal: 100 });
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
|
||||
const fetchCall = vi.mocked(fetch).mock.calls[0];
|
||||
expect(fetchCall[0]).toBe("http://remote:4040/api/project-health?projectId=proj_456");
|
||||
});
|
||||
});
|
||||
|
||||
// ── Node not found / local / no URL ───────────────────────────────────
|
||||
|
||||
it("returns 404 when node is not found", async () => {
|
||||
mockGetNode.mockResolvedValue(undefined);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/missing_node/health");
|
||||
|
||||
expect(res.status).toBe(404);
|
||||
expect(res.body).toEqual({ error: "Node not found" });
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns 400 when node type is local", async () => {
|
||||
const localNode = makeNode({
|
||||
id: "node_local",
|
||||
name: "local",
|
||||
type: "local",
|
||||
});
|
||||
|
||||
mockGetNode.mockResolvedValue(localNode);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_local/projects");
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body).toEqual({ error: "Cannot proxy to local node" });
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns 400 when remote node has no URL configured", async () => {
|
||||
const nodeNoUrl = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: undefined,
|
||||
});
|
||||
|
||||
mockGetNode.mockResolvedValue(nodeNoUrl);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_remote/health");
|
||||
|
||||
expect(res.status).toBe(400);
|
||||
expect(res.body).toEqual({ error: "Node has no URL configured" });
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// ── Auth header injection ─────────────────────────────────────────────
|
||||
|
||||
it("injects Authorization header when node has apiKey", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
apiKey: "secret-key-123",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
makeMockFetchResponse({ status: 200, ok: true, body: [] }),
|
||||
),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
await request(app, "GET", "/api/proxy/node_remote/projects");
|
||||
|
||||
const fetchCall = vi.mocked(fetch).mock.calls[0];
|
||||
expect(fetchCall[1]?.headers).toEqual({
|
||||
Authorization: "Bearer secret-key-123",
|
||||
});
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not inject Authorization header when node has no apiKey", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
apiKey: undefined,
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
makeMockFetchResponse({ status: 200, ok: true, body: [] }),
|
||||
),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
await request(app, "GET", "/api/proxy/node_remote/health");
|
||||
|
||||
const fetchCall = vi.mocked(fetch).mock.calls[0];
|
||||
expect(fetchCall[1]?.headers).toEqual({});
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// ── Network error handling ─────────────────────────────────────────────
|
||||
|
||||
it("returns 502 when fetch throws TypeError (network error)", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockRejectedValue(new TypeError("fetch failed")),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_remote/health");
|
||||
|
||||
expect(res.status).toBe(502);
|
||||
expect(res.body).toEqual({ error: "Remote node unreachable" });
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns 504 when fetch throws AbortError (timeout)", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
const abortError = new Error("The operation was aborted");
|
||||
abortError.name = "AbortError";
|
||||
|
||||
vi.stubGlobal("fetch", vi.fn().mockRejectedValue(abortError));
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_remote/health");
|
||||
|
||||
expect(res.status).toBe(504);
|
||||
expect(res.body).toEqual({ error: "Remote node timeout" });
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// ── Response header filtering ───────────────────────────────────────────
|
||||
|
||||
it("filters hop-by-hop headers from remote response", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
makeMockFetchResponse({
|
||||
status: 200,
|
||||
ok: true,
|
||||
body: { foo: "bar" },
|
||||
headers: {
|
||||
"content-type": "application/json",
|
||||
"transfer-encoding": "chunked",
|
||||
"connection": "keep-alive",
|
||||
},
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_remote/health");
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
// content-type should be forwarded
|
||||
expect(res.headers["content-type"]).toBe("application/json");
|
||||
// transfer-encoding and connection should NOT be forwarded
|
||||
expect(res.headers["transfer-encoding"]).toBeUndefined();
|
||||
expect(res.headers["connection"]).toBeUndefined();
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// ── SSE Proxy Route ───────────────────────────────────────────────────
|
||||
|
||||
describe("GET /api/proxy/:nodeId/events", () => {
|
||||
it("sets correct SSE headers and streams data", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
makeMockFetchResponse({
|
||||
status: 200,
|
||||
ok: true,
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
streamChunks: [
|
||||
'data: {"event":"task:updated"}\n\n',
|
||||
'data: {"event":"task:created"}\n\n',
|
||||
],
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_remote/events");
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.headers["content-type"]).toBe("text/event-stream");
|
||||
expect(res.headers["cache-control"]).toBe("no-cache");
|
||||
expect(res.headers["connection"]).toBe("keep-alive");
|
||||
expect(res.headers["x-accel-buffering"]).toBe("no");
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("forwards projectId query param to remote events endpoint", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
makeMockFetchResponse({
|
||||
status: 200,
|
||||
ok: true,
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
streamChunks: ["data: ok\n\n"],
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
await request(app, "GET", "/api/proxy/node_remote/events?projectId=proj_789");
|
||||
|
||||
const fetchCall = vi.mocked(fetch).mock.calls[0];
|
||||
expect(fetchCall[0]).toBe("http://remote:4040/api/events?projectId=proj_789");
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns 502 when SSE fetch throws TypeError", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockRejectedValue(new TypeError("getaddrinfo ENOTFOUND")),
|
||||
);
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_remote/events");
|
||||
|
||||
expect(res.status).toBe(502);
|
||||
expect(res.body).toEqual({ error: "Remote node unreachable" });
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns 504 when SSE fetch throws AbortError", async () => {
|
||||
const remoteNode = makeNode({
|
||||
id: "node_remote",
|
||||
name: "remote",
|
||||
type: "remote",
|
||||
url: "http://remote:4040",
|
||||
});
|
||||
|
||||
const abortError = new Error("The user abort");
|
||||
abortError.name = "AbortError";
|
||||
|
||||
vi.stubGlobal("fetch", vi.fn().mockRejectedValue(abortError));
|
||||
|
||||
mockGetNode.mockResolvedValue(remoteNode);
|
||||
|
||||
const res = await request(app, "GET", "/api/proxy/node_remote/events");
|
||||
|
||||
expect(res.status).toBe(504);
|
||||
expect(res.body).toEqual({ error: "Remote node timeout" });
|
||||
expect(mockClose).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -1772,6 +1772,128 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
return getOrCreateProjectStore(projectId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Forward an HTTP request to a remote node.
|
||||
* Validates the node exists, is remote, and has a URL, then proxies the request.
|
||||
* Always closes the CentralCore instance in a finally block.
|
||||
*/
|
||||
async function proxyToRemoteNode(
|
||||
req: Request,
|
||||
res: Response,
|
||||
remotePath: string,
|
||||
options?: { timeoutMs?: number },
|
||||
): Promise<void> {
|
||||
const nodeId = req.params.nodeId as string;
|
||||
const timeoutMs = options?.timeoutMs ?? 10_000;
|
||||
|
||||
const { CentralCore } = await import("@fusion/core");
|
||||
const central = new CentralCore(store.getFusionDir());
|
||||
|
||||
try {
|
||||
await central.init();
|
||||
|
||||
const node = await central.getNode(nodeId);
|
||||
if (!node) {
|
||||
throw notFound("Node not found");
|
||||
}
|
||||
|
||||
if (node.type === "local") {
|
||||
throw badRequest("Cannot proxy to local node");
|
||||
}
|
||||
|
||||
if (!node.url) {
|
||||
throw badRequest("Node has no URL configured");
|
||||
}
|
||||
|
||||
// Parse query string from original request URL
|
||||
const parsedUrl = new URL(req.url, "http://localhost");
|
||||
const queryString = parsedUrl.search;
|
||||
|
||||
// Build target URL: node.url + /api + remotePath + queryString
|
||||
const targetPath = `/api${remotePath}${queryString}`;
|
||||
const targetUrl = new URL(targetPath, node.url).toString();
|
||||
|
||||
// Build headers, injecting Authorization if apiKey is present
|
||||
const headers: Record<string, string> = {};
|
||||
if (node.apiKey) {
|
||||
headers["Authorization"] = `Bearer ${node.apiKey}`;
|
||||
}
|
||||
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), timeoutMs);
|
||||
|
||||
const response = await fetch(targetUrl, {
|
||||
headers,
|
||||
signal: controller.signal,
|
||||
});
|
||||
|
||||
clearTimeout(timeout);
|
||||
|
||||
// Filter out hop-by-hop headers that should not be forwarded
|
||||
const hopByHopHeaders = new Set([
|
||||
"transfer-encoding",
|
||||
"connection",
|
||||
"keep-alive",
|
||||
"upgrade",
|
||||
"proxy-authenticate",
|
||||
"proxy-authorization",
|
||||
"te",
|
||||
"trailers",
|
||||
]);
|
||||
|
||||
// Forward safe headers from remote response
|
||||
response.headers.forEach((value, key) => {
|
||||
if (!hopByHopHeaders.has(key.toLowerCase())) {
|
||||
res.setHeader(key, value);
|
||||
}
|
||||
});
|
||||
|
||||
res.status(response.status);
|
||||
|
||||
if (!response.body) {
|
||||
res.end();
|
||||
return;
|
||||
}
|
||||
|
||||
// Convert web ReadableStream to Node Readable and pipe
|
||||
const { Readable } = await import("node:stream");
|
||||
const nodeStream = Readable.fromWeb(response.body as import("node:stream/web").ReadableStream);
|
||||
|
||||
nodeStream.on("data", (chunk: Buffer) => {
|
||||
res.write(chunk);
|
||||
});
|
||||
|
||||
nodeStream.on("end", () => {
|
||||
res.end();
|
||||
});
|
||||
|
||||
nodeStream.on("error", (err: Error) => {
|
||||
// Log but don't crash — stream may already be closing
|
||||
console.error(`[proxy] Stream error for node ${nodeId}:`, err.message);
|
||||
if (!res.writableEnded) {
|
||||
res.end();
|
||||
}
|
||||
});
|
||||
} catch (err: unknown) {
|
||||
// Only send error response if headers haven't been sent yet
|
||||
if (res.headersSent) {
|
||||
return;
|
||||
}
|
||||
if (err instanceof Error && err.name === "AbortError") {
|
||||
res.status(504).json({ error: "Remote node timeout" });
|
||||
} else if (err instanceof TypeError) {
|
||||
// DNS failure, connection refused, etc.
|
||||
res.status(502).json({ error: "Remote node unreachable" });
|
||||
} else if (err instanceof ApiError) {
|
||||
throw err;
|
||||
} else {
|
||||
rethrowAsApiError(err, "Proxy request failed");
|
||||
}
|
||||
} finally {
|
||||
await central.close();
|
||||
}
|
||||
}
|
||||
|
||||
interface ProjectContext {
|
||||
store: TaskStore;
|
||||
engine: import("@fusion/engine").ProjectEngine | undefined;
|
||||
@@ -15298,7 +15420,191 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
}
|
||||
rethrowAsApiError(err, "Failed to fetch skills catalog");
|
||||
}
|
||||
}); return router;
|
||||
});
|
||||
|
||||
// ── Remote Node Proxy Routes ───────────────────────────────────────────
|
||||
|
||||
/** GET /api/proxy/:nodeId/health — Forward health check to remote node */
|
||||
router.get("/proxy/:nodeId/health", async function (req, res) {
|
||||
try {
|
||||
await proxyToRemoteNode(req, res, "/health");
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
}
|
||||
rethrowAsApiError(err);
|
||||
}
|
||||
});
|
||||
|
||||
/** GET /api/proxy/:nodeId/projects — Forward projects list to remote node */
|
||||
router.get("/proxy/:nodeId/projects", async function (req, res) {
|
||||
try {
|
||||
await proxyToRemoteNode(req, res, "/projects");
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
}
|
||||
rethrowAsApiError(err);
|
||||
}
|
||||
});
|
||||
|
||||
/** GET /api/proxy/:nodeId/tasks — Forward tasks list to remote node (forwards projectId, q query params) */
|
||||
router.get("/proxy/:nodeId/tasks", async function (req, res) {
|
||||
try {
|
||||
await proxyToRemoteNode(req, res, "/tasks");
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
}
|
||||
rethrowAsApiError(err);
|
||||
}
|
||||
});
|
||||
|
||||
/** GET /api/proxy/:nodeId/project-health — Forward project health to remote node (forwards projectId query param) */
|
||||
router.get("/proxy/:nodeId/project-health", async function (req, res) {
|
||||
try {
|
||||
await proxyToRemoteNode(req, res, "/project-health");
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
}
|
||||
rethrowAsApiError(err);
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* GET /api/proxy/:nodeId/events — SSE proxy to remote node events stream.
|
||||
* Uses a 30-second timeout since SSE connections are long-lived.
|
||||
* Handles client disconnect gracefully.
|
||||
*/
|
||||
router.get("/proxy/:nodeId/events", async function (req, res) {
|
||||
const nodeId = req.params.nodeId as string;
|
||||
|
||||
const { CentralCore } = await import("@fusion/core");
|
||||
const central = new CentralCore(store.getFusionDir());
|
||||
|
||||
try {
|
||||
await central.init();
|
||||
|
||||
const node = await central.getNode(nodeId);
|
||||
if (!node) {
|
||||
res.status(404).json({ error: "Node not found" });
|
||||
return;
|
||||
}
|
||||
|
||||
if (node.type === "local") {
|
||||
res.status(400).json({ error: "Cannot proxy to local node" });
|
||||
return;
|
||||
}
|
||||
|
||||
if (!node.url) {
|
||||
res.status(400).json({ error: "Node has no URL configured" });
|
||||
return;
|
||||
}
|
||||
|
||||
// Parse query string and build target URL
|
||||
const parsedUrl = new URL(req.url, "http://localhost");
|
||||
const queryString = parsedUrl.search;
|
||||
const targetUrl = new URL(`/api/events${queryString}`, node.url).toString();
|
||||
|
||||
// Build headers, injecting Authorization if apiKey is present
|
||||
const headers: Record<string, string> = {};
|
||||
if (node.apiKey) {
|
||||
headers["Authorization"] = `Bearer ${node.apiKey}`;
|
||||
}
|
||||
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), 30_000);
|
||||
|
||||
const response = await fetch(targetUrl, {
|
||||
headers,
|
||||
signal: controller.signal,
|
||||
});
|
||||
|
||||
clearTimeout(timeout);
|
||||
|
||||
if (!response.ok) {
|
||||
res.status(response.status).json({ error: "Remote node events unavailable" });
|
||||
return;
|
||||
}
|
||||
|
||||
if (!response.body) {
|
||||
res.status(502).json({ error: "Remote node unreachable" });
|
||||
return;
|
||||
}
|
||||
|
||||
// Set SSE headers on Express response
|
||||
res.setHeader("Content-Type", "text/event-stream");
|
||||
res.setHeader("Cache-Control", "no-cache");
|
||||
res.setHeader("Connection", "keep-alive");
|
||||
res.setHeader("X-Accel-Buffering", "no");
|
||||
res.flushHeaders();
|
||||
res.write(": connected\n\n");
|
||||
|
||||
// Convert web ReadableStream to Node Readable and pipe chunks to client
|
||||
const { Readable } = await import("node:stream");
|
||||
const nodeStream = Readable.fromWeb(response.body as import("node:stream/web").ReadableStream);
|
||||
|
||||
let destroyed = false;
|
||||
|
||||
// Handle client disconnect — abort remote fetch and destroy stream
|
||||
req.on("close", () => {
|
||||
if (!destroyed) {
|
||||
destroyed = true;
|
||||
controller.abort();
|
||||
nodeStream.destroy();
|
||||
}
|
||||
});
|
||||
|
||||
nodeStream.on("data", (chunk: Buffer) => {
|
||||
if (!res.writableEnded) {
|
||||
res.write(chunk);
|
||||
}
|
||||
});
|
||||
|
||||
nodeStream.on("end", () => {
|
||||
if (!res.writableEnded) {
|
||||
res.end();
|
||||
}
|
||||
});
|
||||
|
||||
nodeStream.on("error", (err: Error) => {
|
||||
// Log but don't crash — stream may already be closing
|
||||
console.error(`[proxy:sse] Stream error for node ${nodeId}:`, err.message);
|
||||
if (!res.writableEnded) {
|
||||
res.end();
|
||||
}
|
||||
});
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof Error && err.name === "AbortError") {
|
||||
if (!res.headersSent) {
|
||||
res.status(504).json({ error: "Remote node timeout" });
|
||||
} else if (!res.writableEnded) {
|
||||
res.end();
|
||||
}
|
||||
} else if (err instanceof TypeError) {
|
||||
if (!res.headersSent) {
|
||||
res.status(502).json({ error: "Remote node unreachable" });
|
||||
} else if (!res.writableEnded) {
|
||||
res.end();
|
||||
}
|
||||
} else {
|
||||
console.error(`[proxy:sse] Unexpected error for node ${nodeId}:`, err);
|
||||
if (!res.headersSent) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
}
|
||||
rethrowAsApiError(err);
|
||||
} else if (!res.writableEnded) {
|
||||
res.end();
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await central.close();
|
||||
}
|
||||
});
|
||||
|
||||
return router;
|
||||
}
|
||||
|
||||
// ── Automation step helpers ─────────────────────────────────────────
|
||||
@@ -15920,4 +16226,6 @@ function registerAuthRoutes(router: Router, authStorage?: AuthStorageLike): void
|
||||
});
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user