feat(FN-2563): merge fusion/fn-2563 (auto-resolved)
- docs(FN-2563): document proxy registrar ordering and dependencies - fix(FN-2563): remove stale proxy context type import - fix(FN-2563): preserve proxy registrar ordering semantics - feat(FN-2563): complete Step 3 — wire proxy registrar - feat(FN-2563): complete Step 2 — add modular proxy registrar
This commit is contained in:
@@ -56,7 +56,7 @@ import { registerAgentReflectionRatingRoutes } from "./routes/register-agent-ref
|
||||
import { registerAgentImportExportRoutes, registerAgentGenerationRoutes } from "./routes/register-agent-import-export-generation-routes.js";
|
||||
import { registerAgentSkillsRoutes } from "./routes/register-agent-skills-routes.js";
|
||||
import { registerPluginsAutomationRoutes } from "./routes/register-plugins-automation.js";
|
||||
import { registerProxyRoutes } from "./routes/register-proxy.js";
|
||||
import { registerProxyRoutes } from "./routes/register-proxy-routes.js";
|
||||
import { registerModelRoutes } from "./routes/register-model-routes.js";
|
||||
import { registerUsageRoutes } from "./routes/register-usage-routes.js";
|
||||
import { registerAuthRoutes } from "./routes/register-auth-routes.js";
|
||||
@@ -820,7 +820,6 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
router,
|
||||
runtimeLogger,
|
||||
planningLogger,
|
||||
proxyLogger,
|
||||
chatLogger,
|
||||
prioritizeProjectsForCurrentDirectory,
|
||||
getProjectIdFromRequest,
|
||||
@@ -828,7 +827,6 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
getProjectContext,
|
||||
emitRemoteRouteDiagnostic,
|
||||
emitAuthSyncAuditLog,
|
||||
proxyToRemoteNode,
|
||||
parseScopeParam,
|
||||
resolveAutomationStore,
|
||||
resolveRoutineStore,
|
||||
@@ -846,7 +844,6 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
options,
|
||||
runtimeLogger,
|
||||
planningLogger,
|
||||
proxyLogger,
|
||||
chatLogger,
|
||||
prioritizeProjectsForCurrentDirectory,
|
||||
getProjectIdFromRequest,
|
||||
@@ -854,7 +851,6 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
getProjectContext,
|
||||
emitRemoteRouteDiagnostic,
|
||||
emitAuthSyncAuditLog,
|
||||
proxyToRemoteNode,
|
||||
parseScopeParam,
|
||||
resolveAutomationStore,
|
||||
resolveRoutineStore,
|
||||
@@ -902,7 +898,6 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
registerFileWorkspaceRoutes(routeContext);
|
||||
registerAgentsProjectsNodesRoutes(routeContext);
|
||||
registerPluginsAutomationRoutes(routeContext);
|
||||
registerProxyRoutes(routeContext);
|
||||
|
||||
// HeartbeatMonitor for triggering agent execution runs
|
||||
const heartbeatMonitor = options?.heartbeatMonitor;
|
||||
@@ -3741,7 +3736,9 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
|
||||
registerAgentSkillsRoutes(routeContext);
|
||||
|
||||
// Remote node proxy routes are registered by registerProxyRoutes().
|
||||
// Remote node proxy routes stay last so explicit handlers always precede
|
||||
// the wildcard /proxy/:nodeId/{*splat} route in Express match order.
|
||||
registerProxyRoutes(router, { store, runtimeLogger });
|
||||
|
||||
return router;
|
||||
}
|
||||
|
||||
@@ -17,8 +17,8 @@ The context provides core cross-cutting plumbing:
|
||||
- Request/project scoping: `getProjectIdFromRequest`, `getScopedStore`, `getProjectContext`
|
||||
- These are also exported from `context.ts` as canonical helpers for future extraction tasks.
|
||||
- Engine-aware fallback behavior for project-bound and root-store APIs
|
||||
- Runtime loggers and diagnostics emitters (`runtimeLogger`, `planningLogger`, `proxyLogger`, `chatLogger`)
|
||||
- Proxy/auth/audit helpers (`proxyToRemoteNode`, `emitRemoteRouteDiagnostic`, `emitAuthSyncAuditLog`)
|
||||
- Runtime loggers and diagnostics emitters (`runtimeLogger`, `planningLogger`, `chatLogger`)
|
||||
- Proxy/auth/audit helpers (`emitRemoteRouteDiagnostic`, `emitAuthSyncAuditLog`)
|
||||
- Automation/routine resolvers and scope parsing helpers
|
||||
- Shared error normalization (`rethrowAsApiError`)
|
||||
|
||||
@@ -66,7 +66,16 @@ The context provides core cross-cutting plumbing:
|
||||
- `register-agent-import-export-generation-routes.ts` — agent import/export, companies catalog, and `/agents/generate/*` session/spec lifecycle
|
||||
- `register-agent-skills-routes.ts` — skills discovery/content/execution/catalog endpoints coupled to agent capability flow
|
||||
- `register-plugins-automation.ts` — plugin CRUD, automation, routines/webhooks
|
||||
- `register-proxy.ts` — remote-node proxy forwarding and SSE proxy routes
|
||||
- `register-proxy-routes.ts` — remote-node proxy forwarding and SSE proxy routes
|
||||
- Injected dependencies: `{ store, runtimeLogger }`
|
||||
- Endpoint inventory (must remain in this order):
|
||||
1. `GET /proxy/:nodeId/health`
|
||||
2. `GET /proxy/:nodeId/projects`
|
||||
3. `GET /proxy/:nodeId/tasks`
|
||||
4. `GET /proxy/:nodeId/project-health`
|
||||
5. `GET /proxy/:nodeId/events` (SSE pass-through, 30s timeout, client-disconnect cleanup)
|
||||
6. `ALL /proxy/:nodeId/{*splat}` (generic wildcard forwarder)
|
||||
- Shared diagnostics: imports `emitRemoteRouteDiagnostic` and `classifyRemoteRouteError` from `routes/context.ts` so proxy and non-proxy registrars (for example mesh/sync routes) keep one diagnostic classification contract.
|
||||
|
||||
## Ordering rules (critical)
|
||||
|
||||
@@ -76,6 +85,7 @@ Express matches in registration order. Keep registrar and in-registrar route ord
|
||||
2. **Specific operation routes before wildcard paths** (`/files/{*filepath}/copy|move|delete|rename|download|download-zip` before `POST /files/{*filepath}`)
|
||||
- Why: Express route matching is first-win. If the wildcard write route is registered first, paths like `/files/somefolder/delete` will be treated as file writes instead of delete operations.
|
||||
3. **Do not move proxy/script/message/file wildcards ahead of specific routes**
|
||||
- For proxy routes specifically, keep all explicit `GET /proxy/:nodeId/*` handlers ahead of `ALL /proxy/:nodeId/{*splat}` and keep proxy registration last in `createApiRoutes()`.
|
||||
4. **Project/node/sync/discovery ordering constraints must stay intact**:
|
||||
- `/projects/across-nodes` and `/projects/detect` must be registered before `/projects/:id`
|
||||
- `/nodes/:id/settings` must be registered before `/nodes/:id/settings/push|pull|sync-status` and before `/nodes/:id/auth/sync`
|
||||
|
||||
@@ -1,15 +1,17 @@
|
||||
import { Router, type Request, type Response } from "express";
|
||||
import { Router, type Request } from "express";
|
||||
import { resolve, sep } from "node:path";
|
||||
import type { TaskStore } from "@fusion/core";
|
||||
import type { ServerOptions } from "../server.js";
|
||||
import { badRequest, notFound, ApiError, internalError } from "../api-error.js";
|
||||
import { ApiError, internalError } from "../api-error.js";
|
||||
import { getOrCreateProjectStore } from "../project-store-resolver.js";
|
||||
import { createRuntimeLogger } from "../runtime-logger.js";
|
||||
import type { RuntimeLogger } from "../runtime-logger.js";
|
||||
import type {
|
||||
ApiRoutesContext,
|
||||
AuthSyncAuditLogInput,
|
||||
ProjectContext,
|
||||
RemoteRouteDiagnosticInput,
|
||||
RemoteRouteErrorClassification,
|
||||
ScopeValue,
|
||||
} from "./types.js";
|
||||
|
||||
@@ -25,11 +27,7 @@ function rethrowAsApiError(error: unknown, fallbackMessage = "Internal server er
|
||||
throw internalError(fallbackMessage);
|
||||
}
|
||||
|
||||
function classifyRemoteRouteError(error: unknown): {
|
||||
classification: "timeout" | "transport" | "unexpected";
|
||||
errorClass: string;
|
||||
errorMessage: string;
|
||||
} {
|
||||
export function classifyRemoteRouteError(error: unknown): RemoteRouteErrorClassification {
|
||||
const fallbackMessage = String(error);
|
||||
|
||||
if (error instanceof Error) {
|
||||
@@ -108,11 +106,45 @@ export async function getProjectContext(
|
||||
return { store: scopedStore, engine: undefined, projectId };
|
||||
}
|
||||
|
||||
export function emitRemoteRouteDiagnostic(
|
||||
runtimeLogger: RuntimeLogger,
|
||||
input: RemoteRouteDiagnosticInput,
|
||||
): void {
|
||||
const logger = runtimeLogger.child("remote-route").child(input.route);
|
||||
const level = input.level ?? "error";
|
||||
|
||||
const context: Record<string, unknown> = {
|
||||
...(input.nodeId !== undefined ? { nodeId: input.nodeId } : {}),
|
||||
...(input.upstreamPath !== undefined ? { upstreamPath: input.upstreamPath } : {}),
|
||||
...(input.stage !== undefined ? { stage: input.stage } : {}),
|
||||
...(input.operationStage !== undefined ? { operationStage: input.operationStage } : {}),
|
||||
...(input.context ?? {}),
|
||||
};
|
||||
|
||||
if (input.error !== undefined) {
|
||||
const classified = classifyRemoteRouteError(input.error);
|
||||
context.transportClassification = classified.classification;
|
||||
context.errorClass = classified.errorClass;
|
||||
context.errorMessage = classified.errorMessage;
|
||||
}
|
||||
|
||||
if (level === "info") {
|
||||
logger.info(input.message, context);
|
||||
return;
|
||||
}
|
||||
|
||||
if (level === "warn") {
|
||||
logger.warn(input.message, context);
|
||||
return;
|
||||
}
|
||||
|
||||
logger.error(input.message, context);
|
||||
}
|
||||
|
||||
export function createApiRoutesContext(store: TaskStore, options?: ServerOptions): ApiRoutesContext {
|
||||
const router = Router();
|
||||
const runtimeLogger = options?.runtimeLogger?.child("routes") ?? createRuntimeLogger("routes");
|
||||
const planningLogger = runtimeLogger.child("planning");
|
||||
const proxyLogger = runtimeLogger.child("proxy");
|
||||
const chatLogger = runtimeLogger.child("chat");
|
||||
|
||||
function prioritizeProjectsForCurrentDirectory<T extends { path: string }>(projects: T[]): T[] {
|
||||
@@ -141,38 +173,6 @@ export function createApiRoutesContext(store: TaskStore, options?: ServerOptions
|
||||
const resolveScopedStore = (req: Request): Promise<TaskStore> => getScopedStore(req, store);
|
||||
const resolveProjectContext = (req: Request): Promise<ProjectContext> => getProjectContext(req, store, options);
|
||||
|
||||
function emitRemoteRouteDiagnostic(input: RemoteRouteDiagnosticInput): void {
|
||||
const logger = runtimeLogger.child("remote-route").child(input.route);
|
||||
const level = input.level ?? "error";
|
||||
|
||||
const context: Record<string, unknown> = {
|
||||
...(input.nodeId !== undefined ? { nodeId: input.nodeId } : {}),
|
||||
...(input.upstreamPath !== undefined ? { upstreamPath: input.upstreamPath } : {}),
|
||||
...(input.stage !== undefined ? { stage: input.stage } : {}),
|
||||
...(input.operationStage !== undefined ? { operationStage: input.operationStage } : {}),
|
||||
...(input.context ?? {}),
|
||||
};
|
||||
|
||||
if (input.error !== undefined) {
|
||||
const classified = classifyRemoteRouteError(input.error);
|
||||
context.transportClassification = classified.classification;
|
||||
context.errorClass = classified.errorClass;
|
||||
context.errorMessage = classified.errorMessage;
|
||||
}
|
||||
|
||||
if (level === "info") {
|
||||
logger.info(input.message, context);
|
||||
return;
|
||||
}
|
||||
|
||||
if (level === "warn") {
|
||||
logger.warn(input.message, context);
|
||||
return;
|
||||
}
|
||||
|
||||
logger.error(input.message, context);
|
||||
}
|
||||
|
||||
function emitAuthSyncAuditLog(input: AuthSyncAuditLogInput): void {
|
||||
const logger = runtimeLogger.child("settings-sync").child("auth");
|
||||
const level = input.level ?? "info";
|
||||
@@ -201,103 +201,6 @@ export function createApiRoutesContext(store: TaskStore, options?: ServerOptions
|
||||
logger.info("Auth sync diagnostic event", context);
|
||||
}
|
||||
|
||||
async function proxyToRemoteNode(
|
||||
req: Request,
|
||||
res: Response,
|
||||
remotePath: string,
|
||||
proxyOptions?: { timeoutMs?: number },
|
||||
): Promise<void> {
|
||||
const nodeId = req.params.nodeId as string;
|
||||
const timeoutMs = proxyOptions?.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");
|
||||
|
||||
const parsedUrl = new URL(req.url, "http://localhost");
|
||||
const queryString = parsedUrl.search;
|
||||
const targetPath = `/api${remotePath}${queryString}`;
|
||||
const targetUrl = new URL(targetPath, node.url).toString();
|
||||
|
||||
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);
|
||||
|
||||
const hopByHopHeaders = new Set([
|
||||
"transfer-encoding",
|
||||
"connection",
|
||||
"keep-alive",
|
||||
"upgrade",
|
||||
"proxy-authenticate",
|
||||
"proxy-authorization",
|
||||
"te",
|
||||
"trailers",
|
||||
]);
|
||||
|
||||
response.headers.forEach((value, key) => {
|
||||
if (!hopByHopHeaders.has(key.toLowerCase())) {
|
||||
res.setHeader(key, value);
|
||||
}
|
||||
});
|
||||
|
||||
res.status(response.status);
|
||||
|
||||
if (!response.body) {
|
||||
res.end();
|
||||
return;
|
||||
}
|
||||
|
||||
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) => {
|
||||
proxyLogger.error(`Stream error for node ${nodeId}`, { error: err.message });
|
||||
if (!res.writableEnded) {
|
||||
res.end();
|
||||
}
|
||||
});
|
||||
} catch (err: unknown) {
|
||||
if (res.headersSent) {
|
||||
return;
|
||||
}
|
||||
if (err instanceof Error && err.name === "AbortError") {
|
||||
res.status(504).json({ error: "Remote node timeout" });
|
||||
} else if (err instanceof TypeError) {
|
||||
res.status(502).json({ error: "Remote node unreachable" });
|
||||
} else if (err instanceof ApiError) {
|
||||
throw err;
|
||||
} else if (err instanceof Error && err.message) {
|
||||
throw new ApiError(500, err.message);
|
||||
} else {
|
||||
throw new ApiError(500, "Proxy request failed");
|
||||
}
|
||||
} finally {
|
||||
await central.close();
|
||||
}
|
||||
}
|
||||
|
||||
function parseScopeParam(req: Request): ScopeValue | undefined {
|
||||
const rawScope =
|
||||
(typeof req.query.scope === "string" ? req.query.scope : undefined) ??
|
||||
@@ -402,15 +305,13 @@ export function createApiRoutesContext(store: TaskStore, options?: ServerOptions
|
||||
options,
|
||||
runtimeLogger,
|
||||
planningLogger,
|
||||
proxyLogger,
|
||||
chatLogger,
|
||||
prioritizeProjectsForCurrentDirectory,
|
||||
getProjectIdFromRequest,
|
||||
getScopedStore: resolveScopedStore,
|
||||
getProjectContext: resolveProjectContext,
|
||||
emitRemoteRouteDiagnostic,
|
||||
emitRemoteRouteDiagnostic: (input) => emitRemoteRouteDiagnostic(runtimeLogger, input),
|
||||
emitAuthSyncAuditLog,
|
||||
proxyToRemoteNode,
|
||||
parseScopeParam,
|
||||
resolveAutomationStore,
|
||||
resolveRoutineStore,
|
||||
|
||||
@@ -1,14 +1,139 @@
|
||||
import { type Request, type Response } from "express";
|
||||
import { ApiError } from "../api-error.js";
|
||||
import type { ApiRoutesContext } from "./types.js";
|
||||
import { type Request, type Response, type Router } from "express";
|
||||
import type { TaskStore } from "@fusion/core";
|
||||
import type { RuntimeLogger } from "../runtime-logger.js";
|
||||
import { ApiError, badRequest, internalError, notFound } from "../api-error.js";
|
||||
import { emitRemoteRouteDiagnostic } from "./context.js";
|
||||
|
||||
export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
const { router, store, proxyToRemoteNode, emitRemoteRouteDiagnostic, rethrowAsApiError } = ctx;
|
||||
export interface ProxyRoutesDeps {
|
||||
store: TaskStore;
|
||||
runtimeLogger: RuntimeLogger;
|
||||
}
|
||||
|
||||
function rethrowAsApiError(error: unknown, fallbackMessage = "Internal server error"): never {
|
||||
if (error instanceof ApiError) {
|
||||
throw error;
|
||||
}
|
||||
|
||||
if (error instanceof Error) {
|
||||
throw internalError(error.message || fallbackMessage);
|
||||
}
|
||||
|
||||
throw internalError(fallbackMessage);
|
||||
}
|
||||
|
||||
async function proxyToRemoteNode(
|
||||
req: Request,
|
||||
res: Response,
|
||||
remotePath: string,
|
||||
deps: ProxyRoutesDeps,
|
||||
proxyOptions?: { timeoutMs?: number },
|
||||
): Promise<void> {
|
||||
const { store, runtimeLogger } = deps;
|
||||
const proxyLogger = runtimeLogger.child("proxy");
|
||||
const nodeId = req.params.nodeId as string;
|
||||
const timeoutMs = proxyOptions?.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");
|
||||
|
||||
const parsedUrl = new URL(req.url, "http://localhost");
|
||||
const queryString = parsedUrl.search;
|
||||
const targetPath = `/api${remotePath}${queryString}`;
|
||||
const targetUrl = new URL(targetPath, node.url).toString();
|
||||
|
||||
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);
|
||||
|
||||
const hopByHopHeaders = new Set([
|
||||
"transfer-encoding",
|
||||
"connection",
|
||||
"keep-alive",
|
||||
"upgrade",
|
||||
"proxy-authenticate",
|
||||
"proxy-authorization",
|
||||
"te",
|
||||
"trailers",
|
||||
]);
|
||||
|
||||
response.headers.forEach((value, key) => {
|
||||
if (!hopByHopHeaders.has(key.toLowerCase())) {
|
||||
res.setHeader(key, value);
|
||||
}
|
||||
});
|
||||
|
||||
res.status(response.status);
|
||||
|
||||
if (!response.body) {
|
||||
res.end();
|
||||
return;
|
||||
}
|
||||
|
||||
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) => {
|
||||
proxyLogger.error(`Stream error for node ${nodeId}`, { error: err.message });
|
||||
if (!res.writableEnded) {
|
||||
res.end();
|
||||
}
|
||||
});
|
||||
} catch (err: unknown) {
|
||||
if (res.headersSent) {
|
||||
return;
|
||||
}
|
||||
if (err instanceof Error && err.name === "AbortError") {
|
||||
res.status(504).json({ error: "Remote node timeout" });
|
||||
} else if (err instanceof TypeError) {
|
||||
res.status(502).json({ error: "Remote node unreachable" });
|
||||
} else if (err instanceof ApiError) {
|
||||
throw err;
|
||||
} else if (err instanceof Error && err.message) {
|
||||
throw new ApiError(500, err.message);
|
||||
} else {
|
||||
throw new ApiError(500, "Proxy request failed");
|
||||
}
|
||||
} finally {
|
||||
await central.close();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers remote-node proxy forwarding routes.
|
||||
*
|
||||
* Route order is required behavior: specific proxy endpoints must remain ahead of the
|
||||
* wildcard proxy catch-all, and the SSE proxy retains explicit timeout/cleanup semantics.
|
||||
*/
|
||||
export function registerProxyRoutes(router: Router, deps: ProxyRoutesDeps): void {
|
||||
const { store, runtimeLogger } = deps;
|
||||
|
||||
/** 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");
|
||||
await proxyToRemoteNode(req, res, "/health", deps);
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
@@ -20,7 +145,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
/** 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");
|
||||
await proxyToRemoteNode(req, res, "/projects", deps);
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
@@ -32,7 +157,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
/** 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");
|
||||
await proxyToRemoteNode(req, res, "/tasks", deps);
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
@@ -44,7 +169,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
/** 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");
|
||||
await proxyToRemoteNode(req, res, "/project-health", deps);
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
@@ -128,7 +253,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
req.on("close", () => {
|
||||
if (!destroyed) {
|
||||
destroyed = true;
|
||||
emitRemoteRouteDiagnostic({
|
||||
emitRemoteRouteDiagnostic(runtimeLogger, {
|
||||
route: "proxy-sse",
|
||||
message: "Closing SSE proxy stream after client disconnect",
|
||||
nodeId,
|
||||
@@ -154,7 +279,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
});
|
||||
|
||||
nodeStream.on("error", (err: Error) => {
|
||||
emitRemoteRouteDiagnostic({
|
||||
emitRemoteRouteDiagnostic(runtimeLogger, {
|
||||
route: "proxy-sse",
|
||||
message: "SSE proxy stream error",
|
||||
nodeId,
|
||||
@@ -172,7 +297,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
const upstreamPath = `/api/events${queryString}`;
|
||||
|
||||
if (err instanceof Error && err.name === "AbortError") {
|
||||
emitRemoteRouteDiagnostic({
|
||||
emitRemoteRouteDiagnostic(runtimeLogger, {
|
||||
route: "proxy-sse",
|
||||
message: "SSE proxy request timed out",
|
||||
nodeId,
|
||||
@@ -187,7 +312,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
res.end();
|
||||
}
|
||||
} else if (err instanceof TypeError) {
|
||||
emitRemoteRouteDiagnostic({
|
||||
emitRemoteRouteDiagnostic(runtimeLogger, {
|
||||
route: "proxy-sse",
|
||||
message: "SSE proxy transport failure",
|
||||
nodeId,
|
||||
@@ -202,7 +327,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
res.end();
|
||||
}
|
||||
} else {
|
||||
emitRemoteRouteDiagnostic({
|
||||
emitRemoteRouteDiagnostic(runtimeLogger, {
|
||||
route: "proxy-sse",
|
||||
message: "SSE proxy unexpected failure",
|
||||
nodeId,
|
||||
@@ -330,7 +455,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
});
|
||||
|
||||
nodeStream.on("error", (err: Error) => {
|
||||
emitRemoteRouteDiagnostic({
|
||||
emitRemoteRouteDiagnostic(runtimeLogger, {
|
||||
route: "proxy-wildcard",
|
||||
message: "Wildcard proxy stream error",
|
||||
nodeId,
|
||||
@@ -350,7 +475,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
const errorObj = err as { name?: string } | null;
|
||||
const isAbortError = errorObj?.name === "AbortError";
|
||||
if (isAbortError) {
|
||||
emitRemoteRouteDiagnostic({
|
||||
emitRemoteRouteDiagnostic(runtimeLogger, {
|
||||
route: "proxy-wildcard",
|
||||
message: "Wildcard proxy request timed out",
|
||||
nodeId,
|
||||
@@ -364,7 +489,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
}
|
||||
res.status(504).json({ error: "Gateway Timeout" });
|
||||
} else if (err instanceof TypeError) {
|
||||
emitRemoteRouteDiagnostic({
|
||||
emitRemoteRouteDiagnostic(runtimeLogger, {
|
||||
route: "proxy-wildcard",
|
||||
message: "Wildcard proxy transport failure",
|
||||
nodeId,
|
||||
@@ -378,7 +503,7 @@ export function registerProxyRoutes(ctx: ApiRoutesContext): void {
|
||||
}
|
||||
res.status(502).json({ error: "Bad Gateway" });
|
||||
} else {
|
||||
emitRemoteRouteDiagnostic({
|
||||
emitRemoteRouteDiagnostic(runtimeLogger, {
|
||||
route: "proxy-wildcard",
|
||||
message: "Wildcard proxy unexpected failure",
|
||||
nodeId,
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { Request, Response, Router } from "express";
|
||||
import type { Request, Router } from "express";
|
||||
import type { AutomationStore, RoutineStore, TaskStore } from "@fusion/core";
|
||||
import type { ServerOptions } from "../server.js";
|
||||
import type { RuntimeLogger } from "../runtime-logger.js";
|
||||
@@ -21,6 +21,12 @@ export interface RemoteRouteDiagnosticInput {
|
||||
context?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface RemoteRouteErrorClassification {
|
||||
classification: "timeout" | "transport" | "unexpected";
|
||||
errorClass: string;
|
||||
errorMessage: string;
|
||||
}
|
||||
|
||||
export interface AuthSyncAuditLogInput {
|
||||
level?: "info" | "warn" | "error";
|
||||
operation: "receive" | "sync";
|
||||
@@ -39,7 +45,6 @@ export interface ApiRoutesContext {
|
||||
options?: ServerOptions;
|
||||
runtimeLogger: RuntimeLogger;
|
||||
planningLogger: RuntimeLogger;
|
||||
proxyLogger: RuntimeLogger;
|
||||
chatLogger: RuntimeLogger;
|
||||
getProjectIdFromRequest(req: Request): string | undefined;
|
||||
getScopedStore(req: Request): Promise<TaskStore>;
|
||||
@@ -47,7 +52,6 @@ export interface ApiRoutesContext {
|
||||
prioritizeProjectsForCurrentDirectory<T extends { path: string }>(projects: T[]): T[];
|
||||
emitRemoteRouteDiagnostic(input: RemoteRouteDiagnosticInput): void;
|
||||
emitAuthSyncAuditLog(input: AuthSyncAuditLogInput): void;
|
||||
proxyToRemoteNode(req: Request, res: Response, remotePath: string, options?: { timeoutMs?: number }): Promise<void>;
|
||||
parseScopeParam(req: Request): ScopeValue | undefined;
|
||||
resolveAutomationStore(req: Request, scope: ScopeValue | undefined): AutomationStore;
|
||||
resolveRoutineStore(req: Request, scope: ScopeValue | undefined): RoutineStore;
|
||||
|
||||
Reference in New Issue
Block a user