Integrate Inngest for deployment management in the API. Added Inngest client initialization and updated deployment handling to send events to Inngest instead of using Redis. Updated package dependencies to include Inngest version 3.40.1.

apps/api/package.json

@@ -9,6 +9,7 @@
 		"typecheck": "tsc --noEmit"
 	},
 	"dependencies": {
+		"inngest": "3.40.1",
 		"@dokploy/server": "workspace:*",
 		"@hono/node-server": "^1.14.3",
 		"@hono/zod-validator": "0.3.0",

apps/api/src/index.ts

@@ -2,21 +2,79 @@ import { serve } from "@hono/node-server";
 import { Hono } from "hono";
 import "dotenv/config";
 import { zValidator } from "@hono/zod-validator";
-import { Queue } from "@nerimity/mimiqueue";
-import { createClient } from "redis";
+import { Inngest } from "inngest";
+import { serve as serveInngest } from "inngest/hono";
 import { logger } from "./logger.js";
 import { type DeployJob, deployJobSchema } from "./schema.js";
 import { deploy } from "./utils.js";
 
 const app = new Hono();
-const redisClient = createClient({
-	url: process.env.REDIS_URL,
+
+// Initialize Inngest client
+export const inngest = new Inngest({
+	id: "dokploy-deployments",
+	name: "Dokploy Deployment Service",
 });
 
+export const deploymentFunction = inngest.createFunction(
+	{
+		id: "deploy-application",
+		name: "Deploy Application",
+		concurrency: [
+			{
+				key: "event.data.serverId",
+				limit: 1,
+			},
+		],
+		retries: 1,
+	},
+	{ event: "deployment/requested" },
+
+	async ({ event, step }) => {
+		const jobData = event.data as DeployJob;
+
+		return await step.run("execute-deployment", async () => {
+			logger.info("Deploying started");
+
+			try {
+				const result = await deploy(jobData);
+				logger.info("Deployment finished", result);
+
+				// Send success event
+				await inngest.send({
+					name: "deployment/completed",
+					data: {
+						...jobData,
+						result,
+						status: "success",
+					},
+				});
+
+				return result;
+			} catch (error) {
+				logger.error("Deployment failed", { jobData, error });
+
+				// Send failure event
+				await inngest.send({
+					name: "deployment/failed",
+					data: {
+						...jobData,
+						error: error instanceof Error ? error.message : String(error),
+						status: "failed",
+					},
+				});
+
+				throw error;
+			}
+		});
+	},
+);
+
 app.use(async (c, next) => {
-	if (c.req.path === "/health") {
+	if (c.req.path === "/health" || c.req.path === "/api/inngest") {
 		return next();
 	}
+
 	const authHeader = c.req.header("X-API-Key");
 
 	if (process.env.API_KEY !== authHeader) {
@@ -26,36 +84,55 @@ app.use(async (c, next) => {
 	return next();
 });
 
-app.post("/deploy", zValidator("json", deployJobSchema), (c) => {
+app.post("/deploy", zValidator("json", deployJobSchema), async (c) => {
 	const data = c.req.valid("json");
-	queue.add(data, { groupName: data.serverId });
+	logger.info("Received deployment request", data);
+
+	try {
+		// Send event to Inngest instead of adding to Redis queue
+		await inngest.send({
+			name: "deployment/requested",
+			data,
+		});
+
+		logger.info("Deployment event sent to Inngest", {
+			serverId: data.serverId,
+		});
+
 		return c.json(
 			{
-			message: "Deployment Added",
+				message: "Deployment Added to Inngest Queue",
+				serverId: data.serverId,
 			},
 			200,
 		);
+	} catch (error) {
+		console.log("error", error);
+		logger.error("Failed to send deployment event", error);
+		return c.json(
+			{
+				message: "Failed to queue deployment",
+				error: error instanceof Error ? error.message : String(error),
+			},
+			500,
+		);
+	}
 });
 
 app.get("/health", async (c) => {
 	return c.json({ status: "ok" });
 });
 
-const queue = new Queue({
-	name: "deployments",
-	process: async (job: DeployJob) => {
-		logger.info("Deploying job", job);
-		return await deploy(job);
-	},
-	redisClient,
-});
-
-(async () => {
-	await redisClient.connect();
-	await redisClient.flushAll();
-	logger.info("Redis Cleaned");
-})();
+// Serve Inngest functions endpoint
+app.on(
+	["GET", "POST", "PUT"],
+	"/api/inngest",
+	serveInngest({
+		client: inngest,
+		functions: [deploymentFunction],
+	}),
+);
 
 const port = Number.parseInt(process.env.PORT || "3000");
-logger.info("Starting Deployments Server ✅", port);
+logger.info("Starting Deployments Server with Inngest ✅", port);
 serve({ fetch: app.fetch, port });

apps/api/src/utils.ts

@@ -64,7 +64,7 @@ export const deploy = async (job: DeployJob) => {
 				}
 			}
 		}
-	} catch (_) {
+	} catch (e) {
 		if (job.applicationType === "application") {
 			await updateApplicationStatus(job.applicationId, "error");
 		} else if (job.applicationType === "compose") {
@@ -76,6 +76,8 @@ export const deploy = async (job: DeployJob) => {
 				previewStatus: "error",
 			});
 		}
+
+		throw e;
 	}
 
 	return true;