experiment: tasks

packages/api/src/router/cron-jobs.ts

@@ -3,7 +3,6 @@ import z from "zod/v4";
 
 import { createLogger } from "@homarr/core/infrastructure/logs";
 import { cronExpressionSchema, jobGroupKeys, jobNameSchema } from "@homarr/cron-job-api";
-import { cronJobApi } from "@homarr/cron-job-api/client";
 import type { TaskStatus } from "@homarr/cron-job-status";
 import { createCronJobStatusChannel } from "@homarr/cron-job-status";
 
@@ -15,20 +14,20 @@ export const cronJobsRouter = createTRPCRouter({
   triggerJob: permissionRequiredProcedure
     .requiresPermission("admin")
     .input(jobNameSchema)
-    .mutation(async ({ input }) => {
-      await cronJobApi.trigger.mutate(input);
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.triggerAsync(input);
     }),
   startJob: permissionRequiredProcedure
     .requiresPermission("admin")
     .input(jobNameSchema)
-    .mutation(async ({ input }) => {
-      await cronJobApi.start.mutate(input);
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.startAsync(input);
     }),
   stopJob: permissionRequiredProcedure
     .requiresPermission("admin")
     .input(jobNameSchema)
-    .mutation(async ({ input }) => {
-      await cronJobApi.stop.mutate(input);
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.stopAsync(input);
     }),
   updateJobInterval: permissionRequiredProcedure
     .requiresPermission("admin")
@@ -38,23 +37,23 @@ export const cronJobsRouter = createTRPCRouter({
         cron: cronExpressionSchema,
       }),
     )
-    .mutation(async ({ input }) => {
-      await cronJobApi.updateInterval.mutate(input);
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.updateIntervalAsync(input.name, input.cron);
     }),
   disableJob: permissionRequiredProcedure
     .requiresPermission("admin")
     .input(jobNameSchema)
-    .mutation(async ({ input }) => {
-      await cronJobApi.disable.mutate(input);
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.disableAsync(input);
     }),
   enableJob: permissionRequiredProcedure
     .requiresPermission("admin")
     .input(jobNameSchema)
-    .mutation(async ({ input }) => {
-      await cronJobApi.enable.mutate(input);
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.enableAsync(input);
     }),
-  getJobs: permissionRequiredProcedure.requiresPermission("admin").query(async () => {
-    return await cronJobApi.getAll.query();
+  getJobs: permissionRequiredProcedure.requiresPermission("admin").query(async ({ ctx }) => {
+    return await ctx.jobManager.getAllAsync();
   }),
   subscribeToStatusUpdates: permissionRequiredProcedure.requiresPermission("admin").subscription(() => {
     return observable<TaskStatus>((emit) => {

packages/api/src/router/cronJobs.ts

@@ -0,0 +1,59 @@
+import { TRPCError } from "@trpc/server";
+import { z } from "zod/v4";
+
+import { createTRPCRouter, protectedProcedure } from "../trpc";
+import type { JobGroupKeys } from "@homarr/cron-jobs";
+import { jobGroup } from "@homarr/cron-jobs";
+import { cronExpressionSchema } from "@homarr/cron-job-api";
+
+const jobNameSchema = z.enum(jobGroup.getKeys());
+
+export const cronJobsRouter = createTRPCRouter({
+  start: protectedProcedure
+    .meta({ requiredPermission: "admin" })
+    .input(jobNameSchema)
+    .mutation(async ({ input, ctx }) => {
+      // Use the job manager from context
+      await ctx.jobManager.startAsync(input);
+    }),
+  trigger: protectedProcedure
+    .meta({ requiredPermission: "admin" })
+    .input(jobNameSchema)
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.triggerAsync(input);
+    }),
+  stop: protectedProcedure
+    .meta({ requiredPermission: "admin" })
+    .input(jobNameSchema)
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.stopAsync(input);
+    }),
+  updateInterval: protectedProcedure
+    .meta({ requiredPermission: "admin" })
+    .input(
+      z.object({
+        name: jobNameSchema,
+        cron: cronExpressionSchema,
+      }),
+    )
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.updateIntervalAsync(input.name, input.cron);
+    }),
+  disable: protectedProcedure
+    .meta({ requiredPermission: "admin" })
+    .input(jobNameSchema)
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.disableAsync(input);
+    }),
+  enable: protectedProcedure
+    .meta({ requiredPermission: "admin" })
+    .input(jobNameSchema)
+    .mutation(async ({ input, ctx }) => {
+      await ctx.jobManager.enableAsync(input);
+    }),
+  getAll: protectedProcedure
+    .meta({ requiredPermission: "admin" })
+    .query(async ({ ctx }) => {
+      return ctx.jobManager.getAllAsync();
+    }),
+});

packages/api/src/trpc.ts

@@ -34,7 +34,7 @@ const logger = createLogger({ module: "trpc" });
  *
  * @see https://trpc.io/docs/server/context
  */
-export const createTRPCContext = (opts: { headers: Headers; session: Session | null }) => {
+export const createTRPCContext = (opts: { headers: Headers; session: Session | null; jobManager?: any }) => {
   const session = opts.session;
   const source = opts.headers.get("x-trpc-source") ?? "unknown";
 
@@ -44,6 +44,7 @@ export const createTRPCContext = (opts: { headers: Headers; session: Session | n
     session,
     deviceType: userAgent(opts.headers).device.type,
     db,
+    jobManager: opts.jobManager,
   };
 };
 

packages/cron-jobs-core/src/creator.ts

@@ -1,12 +1,9 @@
-import { createTask, validate } from "node-cron";
-
 import { Stopwatch } from "@homarr/common";
 import type { MaybePromise } from "@homarr/common/types";
 import { ErrorWithMetadata } from "@homarr/core/infrastructure/logs/error";
 import { db } from "@homarr/db";
 
 import type { Logger } from "./logger";
-import type { ValidateCron } from "./validation";
 
 export interface CreateCronJobCreatorOptions<TAllowedNames extends string> {
   beforeCallback?: (name: TAllowedNames) => MaybePromise<void>;
@@ -74,28 +71,6 @@ const createCallback = <TAllowedNames extends string, TName extends TAllowedName
     return {
       name,
       cronExpression: defaultCronExpression,
-      async createTaskAsync() {
-        const configuration = await db.query.cronJobConfigurations.findFirst({
-          where: (cronJobConfigurations, { eq }) => eq(cronJobConfigurations.name, name),
-        });
-
-        const cronExpression = options.preventCustomInterval
-          ? defaultCronExpression
-          : (configuration?.cronExpression ?? defaultCronExpression);
-
-        const scheduledTask = createTask(cronExpression, () => void catchingCallbackAsync(), {
-          name,
-          timezone: creatorOptions.timezone,
-        });
-        creatorOptions.logger.logDebug("The scheduled task for cron job was created", {
-          name,
-          cronExpression: defaultCronExpression,
-          timezone: creatorOptions.timezone,
-          runOnStart: options.runOnStart,
-        });
-
-        return scheduledTask;
-      },
       async onStartAsync() {
         if (options.beforeStart) {
           creatorOptions.logger.logDebug("Running beforeStart for job", {
@@ -136,9 +111,7 @@ export const createCronJobCreator = <TAllowedNames extends string = string>(
       name,
       cronExpression: defaultCronExpression,
     });
-    if (!validate(defaultCronExpression)) {
-      throw new Error(`Invalid cron expression '${defaultCronExpression}' for job '${name}'`);
-    }
+    // Validation will be done by cron-parser when needed
     creatorOptions.logger.logDebug("Cron job expression for cron job is valid", {
       name,
       cronExpression: defaultCronExpression,

packages/cron-jobs-core/src/group.ts

@@ -1,11 +1,9 @@
-import type { ScheduledTask } from "node-cron";
-
 import { objectEntries, objectKeys } from "@homarr/common";
 import { db } from "@homarr/db";
+import { jobRegistry } from "./registry";
 
 import type { JobCallback } from "./creator";
 import type { Logger } from "./logger";
-import { jobRegistry } from "./registry";
 
 type Jobs<TAllowedNames extends string> = {
   [name in TAllowedNames]: ReturnType<JobCallback<TAllowedNames, name>>;
@@ -34,86 +32,79 @@ export const createJobGroupCreator = <TAllowedNames extends string = string>(
       });
     }
 
-    const tasks = new Map<string, ScheduledTask>();
-
     return {
       initializeAsync: async () => {
+        // No tasks to create, just ensure jobs are registered
         const configurations = await db.query.cronJobConfigurations.findMany();
-        for (const job of jobRegistry.values()) {
+        for (const job of jobs) {
           const configuration = configurations.find(({ name }) => name === job.name);
           if (configuration?.isEnabled === false) {
             continue;
           }
-          if (tasks.has(job.name)) {
-            continue;
-          }
-
-          const scheduledTask = await job.createTaskAsync();
-
-          tasks.set(job.name, scheduledTask);
+          // Run onStart if needed
+          await job.onStartAsync();
         }
       },
       startAsync: async (name: keyof TJobs) => {
-        const job = jobRegistry.get(name as string);
+        // In new system, start means enable, but since we trigger manually, just log
+        const job = jobs[name];
         if (!job) return;
-        if (!tasks.has(job.name)) return;
 
-        options.logger.logInfo("Starting schedule of cron job.", {
+        options.logger.logInfo("Enabling cron job.", {
           name: job.name,
         });
         await job.onStartAsync();
-        await tasks.get(name as string)?.start();
       },
       startAllAsync: async () => {
-        for (const job of jobRegistry.values()) {
-          if (!tasks.has(job.name)) {
-            continue;
-          }
-
-          options.logger.logInfo("Starting schedule of cron job.", {
+        for (const job of Object.values(jobs)) {
+          options.logger.logInfo("Enabling cron job.", {
             name: job.name,
           });
           await job.onStartAsync();
-          await tasks.get(job.name)?.start();
         }
       },
       runManuallyAsync: async (name: keyof TJobs) => {
-        const job = jobRegistry.get(name as string);
+        const job = jobs[name];
         if (!job) return;
         if (job.preventManualExecution) {
           throw new Error(`The job "${job.name}" can not be executed manually.`);
         }
 
-        options.logger.logInfo("Running schedule cron job manually.", {
+        options.logger.logInfo("Running cron job manually.", {
           name: job.name,
         });
-        await tasks.get(name as string)?.execute();
+        await job.executeAsync();
       },
       stopAsync: async (name: keyof TJobs) => {
-        const job = jobRegistry.get(name as string);
+        // In new system, stop means disable, but since no tasks, just log
+        const job = jobs[name];
         if (!job) return;
 
-        options.logger.logInfo("Stopping schedule of cron job.", {
+        options.logger.logInfo("Disabling cron job.", {
           name: job.name,
         });
-        await tasks.get(name as string)?.stop();
       },
       stopAllAsync: async () => {
-        for (const job of jobRegistry.values()) {
-          options.logger.logInfo("Stopping schedule of cron job.", {
+        for (const job of Object.values(jobs)) {
+          options.logger.logInfo("Disabling cron job.", {
             name: job.name,
           });
-          await tasks.get(job.name)?.stop();
         }
       },
       getJobRegistry() {
-        return jobRegistry as Map<TAllowedNames, ReturnType<JobCallback<TAllowedNames, TAllowedNames>>>;
+        // Return a map of jobs
+        const registry = new Map();
+        for (const [key, job] of objectEntries(jobs)) {
+          registry.set(key, job);
+        }
+        return registry;
       },
-      getTask(name: keyof TJobs) {
-        return tasks.get(name as string) ?? null;
+      getTask() {
+        // No tasks in new system
+        return null;
       },
-      setTask(name: keyof TJobs, task: ScheduledTask) {
-        tasks.set(name as string, task);
+      setTask() {
+        // No tasks
       },
       getKeys() {
         return objectKeys(jobs);

packages/job-scheduler/package.json

@@ -0,0 +1,18 @@
+{
+  "name": "@homarr/job-scheduler",
+  "version": "0.1.0",
+  "private": true,
+  "main": "src/index.ts",
+  "types": "src/index.ts",
+  "dependencies": {
+    "@homarr/common": "workspace:*",
+    "@homarr/core": "workspace:*",
+    "@homarr/cron-job-api": "workspace:*",
+    "@homarr/cron-jobs": "workspace:*",
+    "@homarr/cron-jobs-core": "workspace:*",
+    "@homarr/cron-job-status": "workspace:*",
+    "@homarr/db": "workspace:*",
+    "@homarr/redis": "workspace:*",
+    "cron-parser": "^4.9.0"
+  }
+}

packages/job-scheduler/src/group.ts

@@ -0,0 +1,114 @@
+import { objectEntries, objectKeys } from "@homarr/common";
+import { db } from "@homarr/db";
+import { jobRegistry } from "@homarr/cron-jobs-core";
+
+import type { JobCallback } from "@homarr/cron-jobs-core";
+import type { Logger } from "@homarr/cron-jobs-core";
+
+type Jobs<TAllowedNames extends string> = {
+  [name in TAllowedNames]: ReturnType<JobCallback<TAllowedNames, name>>;
+};
+
+export interface CreateCronJobGroupCreatorOptions {
+  logger: Logger;
+}
+
+export const createJobGroupCreator = <TAllowedNames extends string = string>(
+  options: CreateCronJobGroupCreatorOptions,
+) => {
+  return <TJobs extends Jobs<TAllowedNames>>(jobs: TJobs) => {
+    options.logger.logDebug("Creating job group.", {
+      jobCount: Object.keys(jobs).length,
+    });
+    for (const [key, job] of objectEntries(jobs)) {
+      if (typeof key !== "string" || typeof job.name !== "string") continue;
+
+      options.logger.logDebug("Registering job in the job registry.", {
+        name: job.name,
+      });
+      jobRegistry.set(key, {
+        ...job,
+        name: job.name,
+      });
+    }
+
+    return {
+      initializeAsync: async () => {
+        // No tasks to create, just ensure jobs are registered
+        const configurations = await db.query.cronJobConfigurations.findMany();
+        for (const job of jobs) {
+          const configuration = configurations.find(({ name }) => name === job.name);
+          if (configuration?.isEnabled === false) {
+            continue;
+          }
+          // Run onStart if needed
+          await job.onStartAsync();
+        }
+      },
+      startAsync: async (name: keyof TJobs) => {
+        // In new system, start means enable, but since we trigger manually, just log
+        const job = jobs[name];
+        if (!job) return;
+
+        options.logger.logInfo("Enabling cron job.", {
+          name: job.name,
+        });
+        await job.onStartAsync();
+      },
+      startAllAsync: async () => {
+        for (const job of Object.values(jobs)) {
+          options.logger.logInfo("Enabling cron job.", {
+            name: job.name,
+          });
+          await job.onStartAsync();
+        }
+      },
+      runManuallyAsync: async (name: keyof TJobs) => {
+        const job = jobs[name];
+        if (!job) return;
+        if (job.preventManualExecution) {
+          throw new Error(`The job "${job.name}" can not be executed manually.`);
+        }
+
+        options.logger.logInfo("Running cron job manually.", {
+          name: job.name,
+        });
+        await job.executeAsync();
+      },
+      stopAsync: async (name: keyof TJobs) => {
+        // In new system, stop means disable, but since no tasks, just log
+        const job = jobs[name];
+        if (!job) return;
+
+        options.logger.logInfo("Disabling cron job.", {
+          name: job.name,
+        });
+      },
+      stopAllAsync: async () => {
+        for (const job of Object.values(jobs)) {
+          options.logger.logInfo("Disabling cron job.", {
+            name: job.name,
+          });
+        }
+      },
+      getJobRegistry() {
+        // Return a map of jobs
+        const registry = new Map();
+        for (const [key, job] of objectEntries(jobs)) {
+          registry.set(key, job);
+        }
+        return registry;
+      },
+      getTask() {
+        // No tasks in new system
+        return null;
+      },
+      setTask() {
+        // No tasks
+      },
+      getKeys() {
+        return objectKeys(jobs);
+      },
+    };
+  };
+};

packages/job-scheduler/src/index.ts

@@ -0,0 +1 @@
+export { JobManager } from "./job-manager";

packages/job-scheduler/src/job-manager.ts

@@ -0,0 +1,172 @@
+import { parseExpression } from "cron-parser";
+
+import { createLogger } from "@homarr/core/infrastructure/logs";
+import type { IJobManager } from "@homarr/cron-job-api";
+import type { jobGroup as cronJobGroup, JobGroupKeys } from "@homarr/cron-jobs";
+import type { Database, InferInsertModel } from "@homarr/db";
+import { eq } from "@homarr/db";
+import { cronJobConfigurations } from "@homarr/db/schema";
+
+const logger = createLogger({ module: "jobManager" });
+
+export class JobManager implements IJobManager {
+  private runningJobs = new Set<string>();
+
+  constructor(
+    private db: Database,
+    private jobGroup: typeof cronJobGroup,
+  ) {}
+
+  public async startAsync(name: JobGroupKeys): Promise<void> {
+    // In the new system, start just means enable
+    await this.enableAsync(name);
+  }
+
+  public async triggerAsync(name: JobGroupKeys): Promise<void> {
+    await this.jobGroup.runManuallyAsync(name);
+  }
+
+  public async stopAsync(name: JobGroupKeys): Promise<void> {
+    // In the new system, stop means disable
+    await this.disableAsync(name);
+  }
+
+  public async updateIntervalAsync(name: JobGroupKeys, cron: string): Promise<void> {
+    logger.info("Updating cron job interval", { name, expression: cron });
+    const job = this.jobGroup.getJobRegistry().get(name);
+    if (!job) throw new Error(`Job ${name} not found`);
+
+    // Validate cron expression
+    try {
+      parseExpression(cron);
+    } catch {
+      throw new Error(`Invalid cron expression: ${cron}`);
+    }
+
+    if (job.preventCustomInterval && cron !== job.cronExpression) {
+      throw new Error(`Custom cron expressions are not allowed for job ${name}`);
+    }
+
+    await this.updateConfigurationAsync(name, { cronExpression: cron });
+    logger.info("Cron job interval updated", { name, expression: cron });
+  }
+
+  public async disableAsync(name: JobGroupKeys): Promise<void> {
+    logger.info("Disabling cron job", { name });
+    const job = this.jobGroup.getJobRegistry().get(name);
+    if (!job) throw new Error(`Job ${name} not found`);
+
+    await this.updateConfigurationAsync(name, { isEnabled: false });
+    logger.info("Cron job disabled", { name });
+  }
+
+  public async enableAsync(name: JobGroupKeys): Promise<void> {
+    logger.info("Enabling cron job", { name });
+    await this.updateConfigurationAsync(name, { isEnabled: true });
+    logger.info("Cron job enabled", { name });
+  }
+
+  // New method to trigger jobs based on current time
+  public async triggerJobsAsync(): Promise<void> {
+    const jobs = await this.getAllAsync();
+    const now = new Date();
+
+    for (const jobInfo of jobs) {
+      if (!jobInfo.isEnabled) continue;
+      if (this.runningJobs.has(jobInfo.name)) continue; // Prevent parallel execution
+
+      try {
+        const cronExpression = jobInfo.cron;
+        const interval = parseExpression(cronExpression);
+        const nextRun = interval.next();
+        const prevRun = interval.prev();
+
+        // Check if current time matches the cron (within a small tolerance)
+        const timeDiff = Math.abs(now.getTime() - nextRun.getTime());
+        if (timeDiff < 1000) { // Within 1 second
+          await this.executeJobAsync(jobInfo.name);
+        }
+      } catch (error) {
+        logger.error("Error checking job schedule", { name: jobInfo.name, error });
+      }
+    }
+  }
+
+  private async executeJobAsync(name: string): Promise<void> {
+    if (this.runningJobs.has(name)) return;
+
+    this.runningJobs.add(name);
+    try {
+      logger.info("Executing scheduled job", { name });
+      await this.jobGroup.runManuallyAsync(name as JobGroupKeys);
+      logger.info("Job executed successfully", { name });
+    } catch (error) {
+      logger.error("Job execution failed", { name, error });
+    } finally {
+      this.runningJobs.delete(name);
+    }
+  }
+
+  private async updateConfigurationAsync(
+    name: JobGroupKeys,
+    configuration: Omit<Partial<InferInsertModel<typeof cronJobConfigurations>>, "name">,
+  ) {
+    const existingConfig = await this.db.query.cronJobConfigurations.findFirst({
+      where: (table, { eq }) => eq(table.name, name),
+    });
+
+    logger.debug("Updating cron job configuration", {
+      name,
+      configuration: JSON.stringify(configuration),
+      exists: Boolean(existingConfig),
+    });
+
+    if (existingConfig) {
+      await this.db
+        .update(cronJobConfigurations)
+        .set({ ...configuration, name: undefined })
+        .where(eq(cronJobConfigurations.name, name));
+      logger.debug("Cron job configuration updated", {
+        name,
+        configuration: JSON.stringify(configuration),
+      });
+      return;
+    }
+
+    const job = this.jobGroup.getJobRegistry().get(name);
+    if (!job) throw new Error(`Job ${name} not found`);
+
+    await this.db.insert(cronJobConfigurations).values({
+      name,
+      cronExpression: configuration.cronExpression ?? job.cronExpression,
+      isEnabled: configuration.isEnabled ?? true,
+    });
+    logger.debug("Cron job configuration created", {
+      name,
+      configuration: JSON.stringify(configuration),
+    });
+  }
+
+  public async getAllAsync(): Promise<
+    {
+      name: JobGroupKeys;
+      cron: string;
+      preventManualExecution: boolean;
+      preventCustomInterval: boolean;
+      isEnabled: boolean;
+    }[]
+  > {
+    const configurations = await this.db.query.cronJobConfigurations.findMany();
+
+    return [...this.jobGroup.getJobRegistry().entries()].map(([name, job]) => {
+      const config = configurations.find((config) => config.name === name);
+      return {
+        name,
+        cron: config?.cronExpression ?? job.cronExpression,
+        preventManualExecution: job.preventManualExecution,
+        preventCustomInterval: job.preventCustomInterval,
+        isEnabled: config?.isEnabled ?? true,
+      };
+    });
+  }
+}