diff --git a/Dockerfile b/Dockerfile index a36fbffdb..67290edfb 100644 --- a/Dockerfile +++ b/Dockerfile @@ -43,10 +43,11 @@ RUN mkdir -p /var/cache/nginx && \ COPY --from=builder /app/apps/nextjs/next.config.ts . COPY --from=builder /app/apps/nextjs/package.json . -COPY --from=builder /app/apps/tasks/tasks.cjs ./apps/tasks/tasks.cjs COPY --from=builder /app/apps/websocket/wssServer.cjs ./apps/websocket/wssServer.cjs COPY --from=builder /app/node_modules/better-sqlite3/build/Release/better_sqlite3.node /app/build/better_sqlite3.node +COPY scripts/job-scheduler.js ./scripts/job-scheduler.js + COPY --from=builder /app/packages/db/migrations ./db/migrations # Automatically leverage output traces to reduce image size diff --git a/apps/nextjs/src/app/[locale]/layout.tsx b/apps/nextjs/src/app/[locale]/layout.tsx index 708699bf8..dee683e17 100644 --- a/apps/nextjs/src/app/[locale]/layout.tsx +++ b/apps/nextjs/src/app/[locale]/layout.tsx @@ -26,6 +26,7 @@ import { Analytics } from "~/components/layout/analytics"; import { CrowdinLiveTranslation } from "~/components/layout/crowdin-live-translation"; import { SearchEngineOptimization } from "~/components/layout/search-engine-optimization"; import { getCurrentColorSchemeAsync } from "~/theme/color-scheme"; +import { initializeJobManager } from "~/lib/job-manager"; import { DayJsLoader } from "./_client-providers/dayjs-loader"; import { JotaiProvider } from "./_client-providers/jotai"; import { CustomMantineProvider } from "./_client-providers/mantine"; @@ -83,6 +84,9 @@ export default async function Layout(props: { const colorScheme = await getCurrentColorSchemeAsync(); const direction = isLocaleRTL((await props.params).locale) ? "rtl" : "ltr"; + // Initialize job manager + await initializeJobManager(); + const StackedProvider = composeWrappers([ (innerProps) => { return ; diff --git a/apps/nextjs/src/app/api/internal/trigger-jobs/route.ts b/apps/nextjs/src/app/api/internal/trigger-jobs/route.ts new file mode 100644 index 000000000..23beef399 --- /dev/null +++ b/apps/nextjs/src/app/api/internal/trigger-jobs/route.ts @@ -0,0 +1,20 @@ +import { NextRequest, NextResponse } from "next/server"; + +import { getJobManager } from "~/lib/job-manager"; + +export async function POST(request: NextRequest) { + // Only allow internal requests + const internalHeader = request.headers.get("x-internal-request"); + if (internalHeader !== "true") { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + try { + const jobManager = getJobManager(); + await jobManager.triggerJobsAsync(); + return NextResponse.json({ success: true }); + } catch (error) { + console.error("Error triggering jobs:", error); + return NextResponse.json({ error: "Internal server error" }, { status: 500 }); + } +} \ No newline at end of file diff --git a/apps/nextjs/src/app/api/trpc/[trpc]/route.ts b/apps/nextjs/src/app/api/trpc/[trpc]/route.ts index 8ac6c166d..7166a9fc4 100644 --- a/apps/nextjs/src/app/api/trpc/[trpc]/route.ts +++ b/apps/nextjs/src/app/api/trpc/[trpc]/route.ts @@ -9,6 +9,7 @@ import { ipAddressFromHeaders } from "@homarr/common/server"; import { createLogger } from "@homarr/core/infrastructure/logs"; import { ErrorWithMetadata } from "@homarr/core/infrastructure/logs/error"; import { db } from "@homarr/db"; +import { getJobManager } from "~/lib/job-manager"; const logger = createLogger({ module: "trpcRoute" }); @@ -45,7 +46,7 @@ const handler = auth(async (req) => { endpoint: trpcPath, router: appRouter, req, - createContext: () => createTRPCContext({ session, headers: req.headers }), + createContext: () => createTRPCContext({ session, headers: req.headers, jobManager: getJobManager() }), onError({ error, path, type }) { logger.error(new ErrorWithMetadata("tRPC Error occured", { path, type }, { cause: error })); }, diff --git a/apps/nextjs/src/lib/job-manager.ts b/apps/nextjs/src/lib/job-manager.ts new file mode 100644 index 000000000..1edb13133 --- /dev/null +++ b/apps/nextjs/src/lib/job-manager.ts @@ -0,0 +1,17 @@ +import { JobManager } from "@homarr/job-scheduler"; +import { jobGroup } from "@homarr/cron-jobs"; +import { db } from "@homarr/db"; + +let jobManagerInstance: JobManager | null = null; + +export function getJobManager(): JobManager { + if (!jobManagerInstance) { + jobManagerInstance = new JobManager(db, jobGroup); + } + return jobManagerInstance; +} + +export async function initializeJobManager(): Promise { + const manager = getJobManager(); + await jobGroup.initializeAsync(); +} \ No newline at end of file diff --git a/apps/tasks/eslint.config.js b/apps/tasks/eslint.config.js deleted file mode 100644 index f44e21cd5..000000000 --- a/apps/tasks/eslint.config.js +++ /dev/null @@ -1,9 +0,0 @@ -import baseConfig from "@homarr/eslint-config/base"; - -/** @type {import('typescript-eslint').Config} */ -export default [ - { - ignores: ["tasks.cjs"], - }, - ...baseConfig, -]; diff --git a/apps/tasks/package.json b/apps/tasks/package.json deleted file mode 100644 index dc6f5eb1a..000000000 --- a/apps/tasks/package.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "name": "@homarr/tasks", - "version": "0.1.0", - "private": true, - "license": "Apache-2.0", - "type": "module", - "exports": { - ".": "./src/index.ts" - }, - "main": "./src/main.ts", - "types": "./src/main.ts", - "scripts": { - "build": "esbuild src/main.ts --bundle --platform=node --loader:.scss=text --external:*.node --external:@opentelemetry/api --external:deasync --external:bcrypt --outfile=tasks.cjs", - "clean": "rm -rf .turbo node_modules", - "dev": "pnpm with-env tsx ./src/main.ts", - "format": "prettier --check . --ignore-path ../../.gitignore", - "lint": "eslint", - "typecheck": "tsc --noEmit", - "with-env": "dotenv -e ../../.env --" - }, - "prettier": "@homarr/prettier-config", - "dependencies": { - "@homarr/analytics": "workspace:^0.1.0", - "@homarr/auth": "workspace:^0.1.0", - "@homarr/common": "workspace:^0.1.0", - "@homarr/core": "workspace:^", - "@homarr/cron-job-api": "workspace:^0.1.0", - "@homarr/cron-jobs": "workspace:^0.1.0", - "@homarr/cron-jobs-core": "workspace:^0.1.0", - "@homarr/db": "workspace:^0.1.0", - "@homarr/definitions": "workspace:^0.1.0", - "@homarr/icons": "workspace:^0.1.0", - "@homarr/integrations": "workspace:^0.1.0", - "@homarr/ping": "workspace:^0.1.0", - "@homarr/redis": "workspace:^0.1.0", - "@homarr/request-handler": "workspace:^0.1.0", - "@homarr/server-settings": "workspace:^0.1.0", - "@homarr/validation": "workspace:^0.1.0", - "@homarr/widgets": "workspace:^0.1.0", - "dayjs": "^1.11.20", - "dotenv": "^17.3.1", - "fastify": "^5.8.2", - "superjson": "2.2.6", - "undici": "7.18.2" - }, - "devDependencies": { - "@homarr/eslint-config": "workspace:^0.2.0", - "@homarr/prettier-config": "workspace:^0.1.0", - "@homarr/tsconfig": "workspace:^0.1.0", - "@types/node": "^24.12.0", - "dotenv-cli": "^11.0.0", - "esbuild": "^0.27.4", - "eslint": "^9.39.4", - "prettier": "^3.8.1", - "tsx": "4.20.4", - "typescript": "^5.9.3" - } -} diff --git a/apps/tasks/src/main.ts b/apps/tasks/src/main.ts deleted file mode 100644 index 4c60f1646..000000000 --- a/apps/tasks/src/main.ts +++ /dev/null @@ -1,52 +0,0 @@ -// This import has to be the first import in the file so that the agent is overridden before any other modules are imported. -import "./overrides"; - -import type { FastifyTRPCPluginOptions } from "@trpc/server/adapters/fastify"; -import { fastifyTRPCPlugin } from "@trpc/server/adapters/fastify"; -import fastify from "fastify"; - -import { createLogger } from "@homarr/core/infrastructure/logs"; -import { ErrorWithMetadata } from "@homarr/core/infrastructure/logs/error"; -import type { JobRouter } from "@homarr/cron-job-api"; -import { jobRouter } from "@homarr/cron-job-api"; -import { CRON_JOB_API_KEY_HEADER, CRON_JOB_API_PATH, CRON_JOB_API_PORT } from "@homarr/cron-job-api/constants"; -import { jobGroup } from "@homarr/cron-jobs"; -import { db } from "@homarr/db"; - -import { JobManager } from "./job-manager"; -import { onStartAsync } from "./on-start"; - -const logger = createLogger({ module: "tasksMain" }); - -const server = fastify({ - maxParamLength: 5000, -}); -server.register(fastifyTRPCPlugin, { - prefix: CRON_JOB_API_PATH, - trpcOptions: { - router: jobRouter, - createContext: ({ req }) => ({ - manager: new JobManager(db, jobGroup), - apiKey: req.headers[CRON_JOB_API_KEY_HEADER] as string | undefined, - }), - onError({ path, error }) { - logger.error(new ErrorWithMetadata("Error in tasks tRPC handler", { path }, { cause: error })); - }, - } satisfies FastifyTRPCPluginOptions["trpcOptions"], -}); - -void (async () => { - await onStartAsync(); - await jobGroup.initializeAsync(); - await jobGroup.startAllAsync(); - - try { - await server.listen({ port: CRON_JOB_API_PORT }); - logger.info("Tasks web server started successfully", { port: CRON_JOB_API_PORT }); - } catch (err) { - logger.error( - new ErrorWithMetadata("Failed to start tasks web server", { port: CRON_JOB_API_PORT }, { cause: err }), - ); - process.exit(1); - } -})(); diff --git a/apps/tasks/src/on-start/index.ts b/apps/tasks/src/on-start/index.ts deleted file mode 100644 index cb9564d1b..000000000 --- a/apps/tasks/src/on-start/index.ts +++ /dev/null @@ -1,7 +0,0 @@ -import { invalidateUpdateCheckerCacheAsync } from "./invalidate-update-checker-cache"; -import { cleanupSessionsAsync } from "./session-cleanup"; - -export async function onStartAsync() { - await cleanupSessionsAsync(); - await invalidateUpdateCheckerCacheAsync(); -} diff --git a/apps/tasks/src/on-start/invalidate-update-checker-cache.ts b/apps/tasks/src/on-start/invalidate-update-checker-cache.ts deleted file mode 100644 index 470989dcf..000000000 --- a/apps/tasks/src/on-start/invalidate-update-checker-cache.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { createLogger } from "@homarr/core/infrastructure/logs"; -import { updateCheckerRequestHandler } from "@homarr/request-handler/update-checker"; - -const logger = createLogger({ module: "invalidateUpdateCheckerCache" }); - -/** - * Invalidates the update checker cache on startup to ensure fresh data. - * It is important as we want to avoid showing pending updates after the update to latest version. - */ -export async function invalidateUpdateCheckerCacheAsync() { - try { - const handler = updateCheckerRequestHandler.handler({}); - await handler.invalidateAsync(); - logger.debug("Update checker cache invalidated"); - } catch (error) { - logger.error(new Error("Failed to invalidate update checker cache", { cause: error })); - } -} diff --git a/apps/tasks/src/on-start/session-cleanup.ts b/apps/tasks/src/on-start/session-cleanup.ts deleted file mode 100644 index 3ef6d9cdc..000000000 --- a/apps/tasks/src/on-start/session-cleanup.ts +++ /dev/null @@ -1,41 +0,0 @@ -import { env } from "@homarr/auth/env"; -import { createLogger } from "@homarr/core/infrastructure/logs"; -import { db, eq, inArray } from "@homarr/db"; -import { sessions, users } from "@homarr/db/schema"; -import { supportedAuthProviders } from "@homarr/definitions"; - -const logger = createLogger({ module: "sessionCleanup" }); - -/** - * Deletes sessions for users that have inactive auth providers. - * Sessions from other providers are deleted so they can no longer be used. - */ -export async function cleanupSessionsAsync() { - try { - const currentAuthProviders = env.AUTH_PROVIDERS; - - const inactiveAuthProviders = supportedAuthProviders.filter((provider) => !currentAuthProviders.includes(provider)); - const subQuery = db - .select({ id: users.id }) - .from(users) - .where(inArray(users.provider, inactiveAuthProviders)) - .as("sq"); - const sessionsWithInactiveProviders = await db - .select({ userId: sessions.userId }) - .from(sessions) - .rightJoin(subQuery, eq(sessions.userId, subQuery.id)); - - const userIds = sessionsWithInactiveProviders.map(({ userId }) => userId).filter((value) => value !== null); - await db.delete(sessions).where(inArray(sessions.userId, userIds)); - - if (sessionsWithInactiveProviders.length > 0) { - logger.info("Deleted sessions for inactive providers", { - count: userIds.length, - }); - } else { - logger.debug("No sessions to delete"); - } - } catch (error) { - logger.error(new Error("Failed to clean up sessions", { cause: error })); - } -} diff --git a/apps/tasks/src/overrides.ts b/apps/tasks/src/overrides.ts deleted file mode 100644 index 684531f81..000000000 --- a/apps/tasks/src/overrides.ts +++ /dev/null @@ -1,5 +0,0 @@ -import { setGlobalDispatcher } from "undici"; - -import { UndiciHttpAgent } from "@homarr/core/infrastructure/http"; - -setGlobalDispatcher(new UndiciHttpAgent()); diff --git a/apps/tasks/tsconfig.json b/apps/tasks/tsconfig.json deleted file mode 100644 index 51e14eb17..000000000 --- a/apps/tasks/tsconfig.json +++ /dev/null @@ -1,13 +0,0 @@ -{ - "extends": "@homarr/tsconfig/base.json", - "compilerOptions": { - "baseUrl": ".", - "paths": { - "~/*": ["src/*"], - "@homarr/node-unifi": ["../../node_modules/@types/node-unifi"] - }, - "tsBuildInfoFile": "node_modules/.cache/tsbuildinfo.json" - }, - "include": ["."], - "exclude": ["node_modules"] -} diff --git a/packages/api/src/router/cron-jobs.ts b/packages/api/src/router/cron-jobs.ts index 5428ac374..e361989e1 100644 --- a/packages/api/src/router/cron-jobs.ts +++ b/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((emit) => { diff --git a/packages/api/src/router/cronJobs.ts b/packages/api/src/router/cronJobs.ts new file mode 100644 index 000000000..39159a905 --- /dev/null +++ b/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(); + }), +}); \ No newline at end of file diff --git a/packages/api/src/trpc.ts b/packages/api/src/trpc.ts index bedd60f3d..e16b03d1b 100644 --- a/packages/api/src/trpc.ts +++ b/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, }; }; diff --git a/packages/cron-jobs-core/src/creator.ts b/packages/cron-jobs-core/src/creator.ts index 4f7a9310f..4c1bcfe9a 100644 --- a/packages/cron-jobs-core/src/creator.ts +++ b/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 { beforeCallback?: (name: TAllowedNames) => MaybePromise; @@ -74,28 +71,6 @@ const createCallback = 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 = ( 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, diff --git a/packages/cron-jobs-core/src/group.ts b/packages/cron-jobs-core/src/group.ts index 77cd6057d..5e5f2787f 100644 --- a/packages/cron-jobs-core/src/group.ts +++ b/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 = { [name in TAllowedNames]: ReturnType>; @@ -34,86 +32,79 @@ export const createJobGroupCreator = ( }); } - const tasks = new Map(); - 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>>; + // 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); diff --git a/packages/job-scheduler/package.json b/packages/job-scheduler/package.json new file mode 100644 index 000000000..6e345ed9d --- /dev/null +++ b/packages/job-scheduler/package.json @@ -0,0 +1,19 @@ +{ + "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" + } +} +/workspaces/homarr/packages/job-scheduler/package.json \ No newline at end of file diff --git a/packages/job-scheduler/src/group.ts b/packages/job-scheduler/src/group.ts new file mode 100644 index 000000000..7c3ddddd5 --- /dev/null +++ b/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 = { + [name in TAllowedNames]: ReturnType>; +}; + +export interface CreateCronJobGroupCreatorOptions { + logger: Logger; +} + +export const createJobGroupCreator = ( + options: CreateCronJobGroupCreatorOptions, +) => { + return >(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); + }, + }; + }; +}; \ No newline at end of file diff --git a/packages/job-scheduler/src/index.ts b/packages/job-scheduler/src/index.ts new file mode 100644 index 000000000..d2040dd7c --- /dev/null +++ b/packages/job-scheduler/src/index.ts @@ -0,0 +1 @@ +export { JobManager } from "./job-manager"; \ No newline at end of file diff --git a/apps/tasks/src/job-manager.ts b/packages/job-scheduler/src/job-manager.ts similarity index 68% rename from apps/tasks/src/job-manager.ts rename to packages/job-scheduler/src/job-manager.ts index f4ec8bf97..ccd7c454b 100644 --- a/apps/tasks/src/job-manager.ts +++ b/packages/job-scheduler/src/job-manager.ts @@ -1,4 +1,4 @@ -import { schedule, validate as validateCron } from "node-cron"; +import { parseExpression } from "cron-parser"; import { createLogger } from "@homarr/core/infrastructure/logs"; import type { IJobManager } from "@homarr/cron-job-api"; @@ -10,58 +10,103 @@ import { cronJobConfigurations } from "@homarr/db/schema"; const logger = createLogger({ module: "jobManager" }); export class JobManager implements IJobManager { + private runningJobs = new Set(); + constructor( private db: Database, private jobGroup: typeof cronJobGroup, ) {} public async startAsync(name: JobGroupKeys): Promise { - await this.jobGroup.startAsync(name); + // In the new system, start just means enable + await this.enableAsync(name); } + public async triggerAsync(name: JobGroupKeys): Promise { await this.jobGroup.runManuallyAsync(name); } + public async stopAsync(name: JobGroupKeys): Promise { - await this.jobGroup.stopAsync(name); + // In the new system, stop means disable + await this.disableAsync(name); } + public async updateIntervalAsync(name: JobGroupKeys, cron: string): Promise { 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`); - if (!validateCron(cron)) { + + // 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 }); - await this.jobGroup.getTask(name)?.destroy(); - - this.jobGroup.setTask( - name, - schedule(cron, () => void job.executeAsync(), { - name, - }), - ); logger.info("Cron job interval updated", { name, expression: cron }); } + public async disableAsync(name: JobGroupKeys): Promise { 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 }); - await this.jobGroup.stopAsync(name); logger.info("Cron job disabled", { name }); } + public async enableAsync(name: JobGroupKeys): Promise { logger.info("Enabling cron job", { name }); await this.updateConfigurationAsync(name, { isEnabled: true }); - await this.jobGroup.startAsync(name); logger.info("Cron job enabled", { name }); } + // New method to trigger jobs based on current time + public async triggerJobsAsync(): Promise { + 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 { + 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>, "name">, @@ -79,7 +124,6 @@ export class JobManager implements IJobManager { if (existingConfig) { await this.db .update(cronJobConfigurations) - // prevent updating the name, as it is the primary key .set({ ...configuration, name: undefined }) .where(eq(cronJobConfigurations.name, name)); logger.debug("Cron job configuration updated", { @@ -97,7 +141,7 @@ export class JobManager implements IJobManager { cronExpression: configuration.cronExpression ?? job.cronExpression, isEnabled: configuration.isEnabled ?? true, }); - logger.debug("Cron job configuration updated", { + logger.debug("Cron job configuration created", { name, configuration: JSON.stringify(configuration), }); @@ -125,4 +169,4 @@ export class JobManager implements IJobManager { }; }); } -} +} \ No newline at end of file diff --git a/scripts/job-scheduler.js b/scripts/job-scheduler.js new file mode 100644 index 000000000..7c04c05a5 --- /dev/null +++ b/scripts/job-scheduler.js @@ -0,0 +1,32 @@ +const http = require('http'); + +function triggerJobs() { + const options = { + hostname: 'localhost', + port: 3000, + path: '/api/internal/trigger-jobs', + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'x-internal-request': 'true' + } + }; + + const req = http.request(options, (res) => { + if (res.statusCode !== 200) { + console.error(`Failed to trigger jobs: ${res.statusCode}`); + } + }); + + req.on('error', (err) => { + console.error('Error triggering jobs:', err); + }); + + req.end(); +} + +// Trigger jobs every second +setInterval(triggerJobs, 1000); + +// Initial trigger +triggerJobs(); \ No newline at end of file diff --git a/scripts/run.sh b/scripts/run.sh index 4d90499d5..7e4016db6 100644 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -54,10 +54,17 @@ WSS_PID=$! node apps/nextjs/server.js & NEXTJS_PID=$! +# Wait a bit for NextJS to start +sleep 5 + +# Start job scheduler +node scripts/job-scheduler.js & +SCHEDULER_PID=$! + # Function to handle SIGTERM and shut down services terminate() { echo "Received SIGTERM. Shutting down..." - kill -TERM $NGINX_PID $TASKS_PID $WSS_PID $NEXTJS_PID 2>/dev/null + kill -TERM $NGINX_PID $SCHEDULER_PID $WSS_PID $NEXTJS_PID 2>/dev/null wait # kill redis-server last because of logging of other services and only if $REDIS_PID is set if [ -n "$REDIS_PID" ]; then