merge: HTTP API for orchestration
This commit is contained in:
@@ -1,28 +1,72 @@
|
||||
import { defineCommand } from "citty";
|
||||
import { loadEnv } from "../env.js";
|
||||
import { getLogger } from "../logger.js";
|
||||
import { startHttpServer } from "../server/http.js";
|
||||
import { disconnectPrisma } from "../orchestrator/persist.js";
|
||||
|
||||
export default defineCommand({
|
||||
meta: {
|
||||
name: "serve",
|
||||
description: "Start the Rails orchestrator server (webhook + Discord bot)",
|
||||
description: "Start the Rails orchestrator HTTP server",
|
||||
},
|
||||
async run() {
|
||||
args: {
|
||||
port: {
|
||||
type: "string",
|
||||
alias: "p",
|
||||
description: "HTTP port",
|
||||
default: "",
|
||||
},
|
||||
host: {
|
||||
type: "string",
|
||||
alias: "H",
|
||||
description: "Bind host",
|
||||
default: "0.0.0.0",
|
||||
},
|
||||
config: {
|
||||
type: "string",
|
||||
alias: "c",
|
||||
description: "Path to rails.config.yaml",
|
||||
default: "",
|
||||
},
|
||||
},
|
||||
async run({ args }) {
|
||||
const env = loadEnv();
|
||||
const log = getLogger();
|
||||
|
||||
const port = parseInt(args.port || String(env.RAILS_PORT), 10);
|
||||
|
||||
const { url, close } = await startHttpServer({
|
||||
port,
|
||||
host: args.host ?? "0.0.0.0",
|
||||
...(args.config && { configPath: args.config }),
|
||||
});
|
||||
|
||||
log.info(
|
||||
{ port: env.RAILS_PORT, nodeEnv: env.NODE_ENV },
|
||||
"hanarang-rails starting",
|
||||
{ url, nodeEnv: env.NODE_ENV },
|
||||
"hanarang-rails server ready",
|
||||
);
|
||||
|
||||
// TODO (Sprint 004): Discord bot initialization
|
||||
// TODO (Sprint 004): Gitea webhook HTTP server
|
||||
// For now, just keep the process alive
|
||||
log.info("Orchestrator running. Press Ctrl+C to stop.");
|
||||
// Graceful shutdown
|
||||
const shutdown = async (signal: string) => {
|
||||
log.info({ signal }, "Shutdown requested");
|
||||
try {
|
||||
await close();
|
||||
await disconnectPrisma();
|
||||
} catch (err) {
|
||||
log.error(
|
||||
{ err: err instanceof Error ? err.message : String(err) },
|
||||
"Shutdown error",
|
||||
);
|
||||
}
|
||||
process.exit(0);
|
||||
};
|
||||
|
||||
process.on("SIGINT", () => void shutdown("SIGINT"));
|
||||
process.on("SIGTERM", () => void shutdown("SIGTERM"));
|
||||
|
||||
// Keep alive
|
||||
await new Promise<never>(() => {
|
||||
// keep alive until signal
|
||||
/* block until signal */
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
191
src/server/http.ts
Normal file
191
src/server/http.ts
Normal file
@@ -0,0 +1,191 @@
|
||||
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
|
||||
import { z } from "zod";
|
||||
import {
|
||||
createPipeline,
|
||||
getPipelineState,
|
||||
listPipelines,
|
||||
sendEvent,
|
||||
} from "../orchestrator/persist.js";
|
||||
import { runPipeline } from "../orchestrator/runner.js";
|
||||
import { loadConfig } from "../config/loader.js";
|
||||
import { MockTransport } from "../handoff/mock-transport.js";
|
||||
import type { SisterTransport } from "../handoff/transport.js";
|
||||
import { childLogger } from "../logger.js";
|
||||
|
||||
const log = childLogger({ module: "http-server" });
|
||||
|
||||
const StartRequest = z.object({
|
||||
project: z.string().min(1),
|
||||
requirements: z.string().default(""),
|
||||
mock: z.boolean().default(true),
|
||||
});
|
||||
|
||||
const AbortRequest = z.object({
|
||||
reason: z.string().default("aborted via api"),
|
||||
});
|
||||
|
||||
interface ServerOpts {
|
||||
port: number;
|
||||
host?: string;
|
||||
configPath?: string;
|
||||
}
|
||||
|
||||
export async function startHttpServer(opts: ServerOpts): Promise<{
|
||||
close: () => Promise<void>;
|
||||
url: string;
|
||||
}> {
|
||||
const host = opts.host ?? "0.0.0.0";
|
||||
const config = await loadConfig(opts.configPath);
|
||||
|
||||
// Build transport map (mock for now; real transports wired in follow-up)
|
||||
const transports = new Map<string, SisterTransport>();
|
||||
const mock = new MockTransport();
|
||||
for (const stage of config.pipeline.stages) {
|
||||
transports.set(stage, mock);
|
||||
}
|
||||
|
||||
const server = createServer(async (req, res) => {
|
||||
const url = new URL(req.url ?? "/", `http://${host}`);
|
||||
const path = url.pathname;
|
||||
const method = req.method ?? "GET";
|
||||
|
||||
log.debug({ method, path }, "Incoming request");
|
||||
|
||||
try {
|
||||
// ── Health ──
|
||||
if (method === "GET" && path === "/health") {
|
||||
return sendJson(res, 200, { ok: true, service: "hanarang-rails" });
|
||||
}
|
||||
|
||||
// ── List pipelines ──
|
||||
if (method === "GET" && path === "/pipelines") {
|
||||
const limit = parseInt(url.searchParams.get("limit") ?? "20", 10);
|
||||
const list = await listPipelines({ limit });
|
||||
return sendJson(res, 200, { pipelines: list });
|
||||
}
|
||||
|
||||
// ── Start new pipeline ──
|
||||
if (method === "POST" && path === "/pipelines/start") {
|
||||
const body = await readJson(req);
|
||||
const parsed = StartRequest.safeParse(body);
|
||||
if (!parsed.success) {
|
||||
return sendJson(res, 400, {
|
||||
error: "invalid_request",
|
||||
issues: parsed.error.issues,
|
||||
});
|
||||
}
|
||||
const { project, requirements, mock: useMock } = parsed.data;
|
||||
|
||||
// Only mock mode is wired right now — real transports come in a follow-up.
|
||||
if (!useMock) {
|
||||
return sendJson(res, 501, {
|
||||
error: "not_implemented",
|
||||
message: "Non-mock transport wiring deferred to next iteration.",
|
||||
});
|
||||
}
|
||||
|
||||
// Run pipeline (async, but we await for this simple demo)
|
||||
const result = await runPipeline({
|
||||
projectName: project,
|
||||
requirements,
|
||||
config,
|
||||
transports,
|
||||
});
|
||||
|
||||
return sendJson(res, 201, {
|
||||
pipelineId: result.pipelineId,
|
||||
finalState: result.finalState,
|
||||
transitions: result.transitions,
|
||||
});
|
||||
}
|
||||
|
||||
// ── Get pipeline status ──
|
||||
const statusMatch = path.match(/^\/pipelines\/([^/]+)$/);
|
||||
if (method === "GET" && statusMatch) {
|
||||
const id = statusMatch[1]!;
|
||||
const state = await getPipelineState(id);
|
||||
if (!state) return sendJson(res, 404, { error: "not_found" });
|
||||
return sendJson(res, 200, state);
|
||||
}
|
||||
|
||||
// ── Abort pipeline ──
|
||||
const abortMatch = path.match(/^\/pipelines\/([^/]+)\/abort$/);
|
||||
if (method === "POST" && abortMatch) {
|
||||
const id = abortMatch[1]!;
|
||||
const body = await readJson(req);
|
||||
const parsed = AbortRequest.safeParse(body || {});
|
||||
const reason = parsed.success
|
||||
? parsed.data.reason
|
||||
: "aborted via api";
|
||||
const result = await sendEvent(id, { type: "ABORT", reason });
|
||||
return sendJson(res, 200, { id, state: result.state });
|
||||
}
|
||||
|
||||
// ── Create pipeline without running ──
|
||||
if (method === "POST" && path === "/pipelines") {
|
||||
const body = await readJson(req);
|
||||
const parsed = StartRequest.safeParse(body);
|
||||
if (!parsed.success) {
|
||||
return sendJson(res, 400, {
|
||||
error: "invalid_request",
|
||||
issues: parsed.error.issues,
|
||||
});
|
||||
}
|
||||
const { project, requirements } = parsed.data;
|
||||
const { pipelineId, state } = await createPipeline(project, requirements);
|
||||
return sendJson(res, 201, { pipelineId, state });
|
||||
}
|
||||
|
||||
return sendJson(res, 404, { error: "not_found", path });
|
||||
} catch (err) {
|
||||
log.error(
|
||||
{ err: err instanceof Error ? err.message : String(err) },
|
||||
"Request handler error",
|
||||
);
|
||||
return sendJson(res, 500, {
|
||||
error: "internal_error",
|
||||
message: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
await new Promise<void>((resolveFn) => {
|
||||
server.listen(opts.port, host, () => resolveFn());
|
||||
});
|
||||
|
||||
const url = `http://${host === "0.0.0.0" ? "127.0.0.1" : host}:${opts.port}`;
|
||||
log.info({ url }, "HTTP server listening");
|
||||
|
||||
return {
|
||||
url,
|
||||
async close() {
|
||||
await new Promise<void>((resolveFn, rejectFn) => {
|
||||
server.close((err) => (err ? rejectFn(err) : resolveFn()));
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function readJson(req: IncomingMessage): Promise<unknown> {
|
||||
return new Promise((resolveFn, rejectFn) => {
|
||||
let body = "";
|
||||
req.on("data", (chunk: Buffer) => (body += chunk.toString()));
|
||||
req.on("end", () => {
|
||||
if (!body) return resolveFn({});
|
||||
try {
|
||||
resolveFn(JSON.parse(body));
|
||||
} catch (err) {
|
||||
rejectFn(err);
|
||||
}
|
||||
});
|
||||
req.on("error", rejectFn);
|
||||
});
|
||||
}
|
||||
|
||||
function sendJson(res: ServerResponse, status: number, body: unknown): void {
|
||||
res.writeHead(status, {
|
||||
"content-type": "application/json",
|
||||
"cache-control": "no-store",
|
||||
});
|
||||
res.end(JSON.stringify(body));
|
||||
}
|
||||
Reference in New Issue
Block a user