refactor: migration

apps/mig/server/queues/deployments-queue.ts

@@ -0,0 +1,138 @@
+import {
+	deployApplication,
+	deployCompose,
+	deployRemoteApplication,
+	deployRemoteCompose,
+	rebuildApplication,
+	rebuildCompose,
+	rebuildRemoteApplication,
+	rebuildRemoteCompose,
+	updateApplicationStatus,
+	updateCompose,
+} from "@dokploy/server";
+import { type Job, Worker } from "bullmq";
+import { myQueue, redisConfig } from "./queueSetup";
+
+type DeployJob =
+	| {
+			applicationId: string;
+			titleLog: string;
+			descriptionLog: string;
+			server?: boolean;
+			type: "deploy" | "redeploy";
+			applicationType: "application";
+			serverId?: string;
+	  }
+	| {
+			composeId: string;
+			titleLog: string;
+			descriptionLog: string;
+			server?: boolean;
+			type: "deploy" | "redeploy";
+			applicationType: "compose";
+			serverId?: string;
+	  };
+
+export type DeploymentJob = DeployJob;
+
+export const deploymentWorker = new Worker(
+	"deployments",
+	async (job: Job<DeploymentJob>) => {
+		try {
+			if (job.data.applicationType === "application") {
+				await updateApplicationStatus(job.data.applicationId, "running");
+				if (job.data.server) {
+					if (job.data.type === "redeploy") {
+						await rebuildRemoteApplication({
+							applicationId: job.data.applicationId,
+							titleLog: job.data.titleLog,
+							descriptionLog: job.data.descriptionLog,
+						});
+					} else if (job.data.type === "deploy") {
+						await deployRemoteApplication({
+							applicationId: job.data.applicationId,
+							titleLog: job.data.titleLog,
+							descriptionLog: job.data.descriptionLog,
+						});
+					}
+				} else {
+					if (job.data.type === "redeploy") {
+						await rebuildApplication({
+							applicationId: job.data.applicationId,
+							titleLog: job.data.titleLog,
+							descriptionLog: job.data.descriptionLog,
+						});
+					} else if (job.data.type === "deploy") {
+						await deployApplication({
+							applicationId: job.data.applicationId,
+							titleLog: job.data.titleLog,
+							descriptionLog: job.data.descriptionLog,
+						});
+					}
+				}
+			} else if (job.data.applicationType === "compose") {
+				await updateCompose(job.data.composeId, {
+					composeStatus: "running",
+				});
+
+				if (job.data.server) {
+					if (job.data.type === "redeploy") {
+						await rebuildRemoteCompose({
+							composeId: job.data.composeId,
+							titleLog: job.data.titleLog,
+							descriptionLog: job.data.descriptionLog,
+						});
+					} else if (job.data.type === "deploy") {
+						await deployRemoteCompose({
+							composeId: job.data.composeId,
+							titleLog: job.data.titleLog,
+							descriptionLog: job.data.descriptionLog,
+						});
+					}
+				} else {
+					if (job.data.type === "deploy") {
+						await deployCompose({
+							composeId: job.data.composeId,
+							titleLog: job.data.titleLog,
+							descriptionLog: job.data.descriptionLog,
+						});
+					} else if (job.data.type === "redeploy") {
+						await rebuildCompose({
+							composeId: job.data.composeId,
+							titleLog: job.data.titleLog,
+							descriptionLog: job.data.descriptionLog,
+						});
+					}
+				}
+			}
+		} catch (error) {
+			console.log("Error", error);
+		}
+	},
+	{
+		autorun: false,
+		connection: redisConfig,
+	},
+);
+
+export const cleanQueuesByApplication = async (applicationId: string) => {
+	const jobs = await myQueue.getJobs(["waiting", "delayed"]);
+
+	for (const job of jobs) {
+		if (job?.data?.applicationId === applicationId) {
+			await job.remove();
+			console.log(`Removed job ${job.id} for application ${applicationId}`);
+		}
+	}
+};
+
+export const cleanQueuesByCompose = async (composeId: string) => {
+	const jobs = await myQueue.getJobs(["waiting", "delayed"]);
+
+	for (const job of jobs) {
+		if (job?.data?.composeId === composeId) {
+			await job.remove();
+			console.log(`Removed job ${job.id} for compose ${composeId}`);
+		}
+	}
+};

apps/mig/server/queues/queueSetup.ts

@@ -0,0 +1,24 @@
+import { type ConnectionOptions, Queue } from "bullmq";
+
+export const redisConfig: ConnectionOptions = {
+	host: process.env.NODE_ENV === "production" ? "dokploy-redis" : "127.0.0.1",
+};
+const myQueue = new Queue("deployments", {
+	connection: redisConfig,
+});
+
+process.on("SIGTERM", () => {
+	myQueue.close();
+	process.exit(0);
+});
+
+myQueue.on("error", (error) => {
+	if ((error as any).code === "ECONNREFUSED") {
+		console.error(
+			"Make sure you have installed Redis and it is running.",
+			error,
+		);
+	}
+});
+
+export { myQueue };