Merge branch 'canary' into 187-backups-for-docker-compose

apps/schedules/src/index.ts

@@ -34,8 +34,8 @@ app.use(async (c, next) => {
 app.post("/create-backup", zValidator("json", jobQueueSchema), async (c) => {
 	const data = c.req.valid("json");
 	scheduleJob(data);
-	logger.info({ data }, "Backup created successfully");
-	return c.json({ message: "Backup created successfully" });
+	logger.info({ data }, `[${data.type}]  created successfully`);
+	return c.json({ message: `[${data.type}]  created successfully` });
 });
 
 app.post("/update-backup", zValidator("json", jobQueueSchema), async (c) => {
@@ -55,6 +55,12 @@ app.post("/update-backup", zValidator("json", jobQueueSchema), async (c) => {
 				type: "server",
 				cronSchedule: job.pattern,
 			});
+		} else if (data.type === "schedule") {
+			result = await removeJob({
+				scheduleId: data.scheduleId,
+				type: "schedule",
+				cronSchedule: job.pattern,
+			});
 		}
 		logger.info({ result }, "Job removed");
 	}

apps/schedules/src/queue.ts

@@ -36,6 +36,12 @@ export const scheduleJob = (job: QueueJob) => {
 				pattern: job.cronSchedule,
 			},
 		});
+	} else if (job.type === "schedule") {
+		jobQueue.add(job.scheduleId, job, {
+			repeat: {
+				pattern: job.cronSchedule,
+			},
+		});
 	}
 };
 
@@ -54,7 +60,13 @@ export const removeJob = async (data: QueueJob) => {
 		});
 		return result;
 	}
-
+	if (data.type === "schedule") {
+		const { scheduleId, cronSchedule } = data;
+		const result = await jobQueue.removeRepeatable(scheduleId, {
+			pattern: cronSchedule,
+		});
+		return result;
+	}
 	return false;
 };
 
@@ -72,6 +84,11 @@ export const getJobRepeatable = async (
 		const job = repeatableJobs.find((j) => j.name === `${serverId}-cleanup`);
 		return job ? job : null;
 	}
+	if (data.type === "schedule") {
+		const { scheduleId } = data;
+		const job = repeatableJobs.find((j) => j.name === scheduleId);
+		return job ? job : null;
+	}
 
 	return null;
 };

apps/schedules/src/schema.ts

@@ -11,6 +11,11 @@ export const jobQueueSchema = z.discriminatedUnion("type", [
 		type: z.literal("server"),
 		serverId: z.string(),
 	}),
+	z.object({
+		cronSchedule: z.string(),
+		type: z.literal("schedule"),
+		scheduleId: z.string(),
+	}),
 ]);
 
 export type QueueJob = z.infer<typeof jobQueueSchema>;

apps/schedules/src/utils.ts

@@ -3,8 +3,10 @@ import {
 	cleanUpSystemPrune,
 	cleanUpUnusedImages,
 	findBackupById,
+	findScheduleById,
 	findServerById,
 	keepLatestNBackups,
+	runCommand,
 	runMariadbBackup,
 	runMongoBackup,
 	runMySqlBackup,
@@ -12,7 +14,7 @@ import {
 	runComposeBackup,
 } from "@dokploy/server";
 import { db } from "@dokploy/server/dist/db";
-import { backups, server } from "@dokploy/server/dist/db/schema";
+import { backups, schedules, server } from "@dokploy/server/dist/db/schema";
 import { and, eq } from "drizzle-orm";
 import { logger } from "./logger.js";
 import { scheduleJob } from "./queue.js";
@@ -75,8 +77,7 @@ export const runJobs = async (job: QueueJob) => {
 				}
 				await runComposeBackup(compose, backup);
 			}
-		}
-		if (job.type === "server") {
+		} else if (job.type === "server") {
 			const { serverId } = job;
 			const server = await findServerById(serverId);
 			if (server.serverStatus === "inactive") {
@@ -86,6 +87,12 @@ export const runJobs = async (job: QueueJob) => {
 			await cleanUpUnusedImages(serverId);
 			await cleanUpDockerBuilder(serverId);
 			await cleanUpSystemPrune(serverId);
+		} else if (job.type === "schedule") {
+			const { scheduleId } = job;
+			const schedule = await findScheduleById(scheduleId);
+			if (schedule.enabled) {
+				await runCommand(schedule.scheduleId);
+			}
 		}
 	} catch (error) {
 		logger.error(error);
@@ -134,4 +141,17 @@ export const initializeJobs = async () => {
 		});
 	}
 	logger.info({ Quantity: backupsResult.length }, "Backups Initialized");
+
+	const schedulesResult = await db.query.schedules.findMany({
+		where: eq(schedules.enabled, true),
+	});
+
+	for (const schedule of schedulesResult) {
+		scheduleJob({
+			scheduleId: schedule.scheduleId,
+			type: "schedule",
+			cronSchedule: schedule.cronExpression,
+		});
+	}
+	logger.info({ Quantity: schedulesResult.length }, "Schedules Initialized");
 };