mirror of
https://github.com/Dokploy/dokploy.git
synced 2026-06-15 20:25:23 +02:00
feat: implement deployment jobs API and enhance queue management
Added a new endpoint to fetch deployment jobs for a server, integrating with the Inngest API to retrieve job details. Updated the queue management system to support centralized job retrieval for cloud environments, improving the deployment monitoring experience. Enhanced the UI to include action buttons for job cancellation and improved error handling for job fetching.
This commit is contained in:
Mauricio Siu
@@ -10,6 +10,7 @@ import {
|
||||
type DeployJob,
|
||||
deployJobSchema,
|
||||
} from "./schema.js";
|
||||
import { fetchDeploymentJobs } from "./service.js";
|
||||
import { deploy } from "./utils.js";
|
||||
|
||||
const app = new Hono();
|
||||
@@ -118,7 +119,6 @@ app.post("/deploy", zValidator("json", deployJobSchema), async (c) => {
|
||||
200,
|
||||
);
|
||||
} catch (error) {
|
||||
console.log("error", error);
|
||||
logger.error("Failed to send deployment event", error);
|
||||
return c.json(
|
||||
{
|
||||
@@ -176,6 +176,29 @@ app.get("/health", async (c) => {
|
||||
return c.json({ status: "ok" });
|
||||
});
|
||||
|
||||
// List deployment jobs (Inngest runs) for a server - same shape as BullMQ queue for the UI
|
||||
app.get("/jobs", async (c) => {
|
||||
const serverId = c.req.query("serverId");
|
||||
if (!serverId) {
|
||||
return c.json({ message: "serverId is required" }, 400);
|
||||
}
|
||||
|
||||
try {
|
||||
const rows = await fetchDeploymentJobs(serverId);
|
||||
return c.json(rows);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
if (message.includes("INNGEST_BASE_URL")) {
|
||||
return c.json(
|
||||
{ message: "INNGEST_BASE_URL is required to list deployment jobs" },
|
||||
503,
|
||||
);
|
||||
}
|
||||
logger.error("Failed to fetch jobs from Inngest", { serverId, error });
|
||||
return c.json([], 200);
|
||||
}
|
||||
});
|
||||
|
||||
// Serve Inngest functions endpoint
|
||||
app.on(
|
||||
["GET", "POST", "PUT"],
|
||||
|
||||
240
apps/api/src/service.ts
Normal file
240
apps/api/src/service.ts
Normal file
@@ -0,0 +1,240 @@
|
||||
import { logger } from "./logger";
|
||||
|
||||
const baseUrl = process.env.INNGEST_BASE_URL ?? "";
|
||||
const signingKey = process.env.INNGEST_SIGNING_KEY ?? "";
|
||||
|
||||
/** Event shape from GET /v1/events (https://api.inngest.com/v1/events) */
|
||||
type InngestEventRow = {
|
||||
internal_id?: string;
|
||||
accountID?: string;
|
||||
environmentID?: string;
|
||||
source?: string;
|
||||
sourceID?: string | null;
|
||||
/** RFC3339 timestamp – API uses receivedAt, dev server may use received_at */
|
||||
receivedAt?: string;
|
||||
received_at?: string;
|
||||
id: string;
|
||||
name: string;
|
||||
data: Record<string, unknown>;
|
||||
user?: unknown;
|
||||
ts: number;
|
||||
v?: string | null;
|
||||
metadata?: {
|
||||
fetchedAt: string;
|
||||
cachedUntil: string | null;
|
||||
};
|
||||
};
|
||||
|
||||
/** Run shape from GET /v1/events/{eventId}/runs – the actual job execution */
|
||||
type InngestRun = {
|
||||
run_id: string;
|
||||
event_id: string;
|
||||
status: string; // "Running" | "Completed" | "Failed" | "Cancelled" | "Queued"?
|
||||
run_started_at?: string;
|
||||
ended_at?: string | null;
|
||||
output?: unknown;
|
||||
// dev server / API may use different casing
|
||||
run_started_at_ms?: number;
|
||||
};
|
||||
|
||||
function getEventReceivedAt(ev: InngestEventRow): string | undefined {
|
||||
return ev.receivedAt ?? ev.received_at;
|
||||
}
|
||||
|
||||
/** Map Inngest run status to BullMQ-style state for the UI */
|
||||
function runStatusToState(
|
||||
status: string,
|
||||
): "pending" | "active" | "completed" | "failed" | "cancelled" {
|
||||
const s = status.toLowerCase();
|
||||
if (s === "running") return "active";
|
||||
if (s === "completed") return "completed";
|
||||
if (s === "failed") return "failed";
|
||||
if (s === "cancelled") return "cancelled";
|
||||
if (s === "queued") return "pending";
|
||||
return "pending";
|
||||
}
|
||||
|
||||
export const fetchInngestEvents = async () => {
|
||||
const maxEvents = MAX_EVENTS;
|
||||
const all: InngestEventRow[] = [];
|
||||
let cursor: string | undefined;
|
||||
|
||||
do {
|
||||
const params = new URLSearchParams({ limit: "100" });
|
||||
if (cursor) {
|
||||
params.set("cursor", cursor);
|
||||
}
|
||||
|
||||
const res = await fetch(`${baseUrl}/v1/events?${params}`, {
|
||||
headers: {
|
||||
Authorization: `Bearer ${signingKey}`,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
});
|
||||
|
||||
if (!res.ok) {
|
||||
logger.warn("Inngest API error", {
|
||||
status: res.status,
|
||||
body: await res.text(),
|
||||
});
|
||||
break;
|
||||
}
|
||||
|
||||
const body = (await res.json()) as {
|
||||
data?: InngestEventRow[];
|
||||
cursor?: string;
|
||||
nextCursor?: string;
|
||||
};
|
||||
const data = Array.isArray(body.data) ? body.data : [];
|
||||
all.push(...data);
|
||||
|
||||
// Next page: API may return cursor/nextCursor, or use last event's internal_id (per API docs)
|
||||
const nextCursor =
|
||||
body.cursor ?? body.nextCursor ?? data[data.length - 1]?.internal_id;
|
||||
const hasMore = data.length === 100 && nextCursor && all.length < maxEvents;
|
||||
cursor = hasMore ? nextCursor : undefined;
|
||||
} while (cursor);
|
||||
|
||||
return all.slice(0, maxEvents);
|
||||
};
|
||||
|
||||
/** Fetch runs for a single event (GET /v1/events/{eventId}/runs) – runs are the actual jobs */
|
||||
export const fetchInngestRunsForEvent = async (
|
||||
eventId: string,
|
||||
): Promise<InngestRun[]> => {
|
||||
const res = await fetch(
|
||||
`${baseUrl}/v1/events/${encodeURIComponent(eventId)}/runs`,
|
||||
{
|
||||
headers: {
|
||||
Authorization: `Bearer ${signingKey}`,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
},
|
||||
);
|
||||
if (!res.ok) {
|
||||
logger.warn("Inngest runs API error", {
|
||||
eventId,
|
||||
status: res.status,
|
||||
body: await res.text(),
|
||||
});
|
||||
return [];
|
||||
}
|
||||
const body = (await res.json()) as { data?: InngestRun[] };
|
||||
return Array.isArray(body.data) ? body.data : [];
|
||||
};
|
||||
|
||||
/** One row for the queue UI (BullMQ-compatible shape) */
|
||||
export type DeploymentJobRow = {
|
||||
id: string;
|
||||
name: string;
|
||||
data: Record<string, unknown>;
|
||||
timestamp: number;
|
||||
processedOn?: number;
|
||||
finishedOn?: number;
|
||||
failedReason?: string;
|
||||
state: string;
|
||||
};
|
||||
|
||||
/** Build queue rows from events + their runs (one row per run, or pending if no run yet) */
|
||||
function buildDeploymentRowsFromRuns(
|
||||
events: InngestEventRow[],
|
||||
runsByEventId: Map<string, InngestRun[]>,
|
||||
serverId: string,
|
||||
): DeploymentJobRow[] {
|
||||
const requested = events.filter(
|
||||
(e) =>
|
||||
e.name === "deployment/requested" &&
|
||||
(e.data as Record<string, unknown>)?.serverId === serverId,
|
||||
);
|
||||
const rows: DeploymentJobRow[] = [];
|
||||
|
||||
for (const ev of requested) {
|
||||
const data = (ev.data ?? {}) as Record<string, unknown>;
|
||||
const runs = runsByEventId.get(ev.id) ?? [];
|
||||
|
||||
if (runs.length === 0) {
|
||||
// Queued: event received but no run yet
|
||||
rows.push({
|
||||
id: ev.id,
|
||||
name: ev.name,
|
||||
data,
|
||||
timestamp: ev.ts,
|
||||
processedOn: ev.ts,
|
||||
finishedOn: undefined,
|
||||
failedReason: undefined,
|
||||
state: "pending",
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const run of runs) {
|
||||
const state = runStatusToState(run.status);
|
||||
const runStartedMs =
|
||||
run.run_started_at_ms ??
|
||||
(run.run_started_at ? new Date(run.run_started_at).getTime() : ev.ts);
|
||||
const endedMs = run.ended_at
|
||||
? new Date(run.ended_at).getTime()
|
||||
: undefined;
|
||||
const failedReason =
|
||||
state === "failed" &&
|
||||
run.output &&
|
||||
typeof run.output === "object" &&
|
||||
"error" in run.output
|
||||
? String((run.output as { error?: unknown }).error)
|
||||
: undefined;
|
||||
|
||||
rows.push({
|
||||
id: run.run_id,
|
||||
name: ev.name,
|
||||
data,
|
||||
timestamp: runStartedMs,
|
||||
processedOn: runStartedMs,
|
||||
finishedOn:
|
||||
state === "completed" || state === "failed" || state === "cancelled"
|
||||
? endedMs
|
||||
: undefined,
|
||||
failedReason,
|
||||
state,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return rows.sort((a, b) => (b.timestamp ?? 0) - (a.timestamp ?? 0));
|
||||
}
|
||||
|
||||
/** Fetch deployment jobs for a server: events → runs → rows (correct model: runs = jobs) */
|
||||
export const fetchDeploymentJobs = async (
|
||||
serverId: string,
|
||||
): Promise<DeploymentJobRow[]> => {
|
||||
if (!signingKey) {
|
||||
logger.warn("INNGEST_SIGNING_KEY not set, returning empty jobs list");
|
||||
return [];
|
||||
}
|
||||
if (!baseUrl) {
|
||||
throw new Error("INNGEST_BASE_URL is required to list deployment jobs");
|
||||
}
|
||||
|
||||
const events = await fetchInngestEvents();
|
||||
|
||||
const requestedForServer = events.filter(
|
||||
(e) =>
|
||||
e.name === "deployment/requested" &&
|
||||
(e.data as Record<string, unknown>)?.serverId === serverId,
|
||||
);
|
||||
// Limit to avoid too many run fetches
|
||||
const toFetch = requestedForServer.slice(0, 50);
|
||||
const runsByEventId = new Map<string, InngestRun[]>();
|
||||
|
||||
await Promise.all(
|
||||
toFetch.map(async (ev) => {
|
||||
const runs = await fetchInngestRunsForEvent(ev.id);
|
||||
runsByEventId.set(ev.id, runs);
|
||||
}),
|
||||
);
|
||||
|
||||
return buildDeploymentRowsFromRuns(events, runsByEventId, serverId);
|
||||
};
|
||||
|
||||
const DEFAULT_MAX_EVENTS = 500;
|
||||
|
||||
const MAX_EVENTS = DEFAULT_MAX_EVENTS;
|
||||
Reference in New Issue
Block a user