feat: integrate tasks/cron server into custom Next.js server

- Add tasks dependencies (fastify, cron-job-api, cron-jobs, etc.) to nextjs app
- Import and start tasks (JobManager, jobGroup, Fastify cron API) in server.ts
- Update root package.json start script to remove tasks.cjs
- Update scripts/run.sh to run single process
- Update Dockerfile to not copy tasks.cjs
- Disable tasks dev script (no-op)

Co-authored-by: ajnart <49837342+ajnart@users.noreply.github.com>

Dockerfile

@@ -43,7 +43,6 @@ 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/nextjs/customServer.cjs ./apps/nextjs/customServer.cjs
 COPY --from=builder /app/node_modules/better-sqlite3/build/Release/better_sqlite3.node /app/build/better_sqlite3.node
 

apps/nextjs/package.json

@@ -5,7 +5,7 @@
   "type": "module",
   "scripts": {
     "build": "pnpm with-env next build && pnpm build:custom-server",
-    "build:custom-server": "esbuild server.ts --bundle --platform=node --outfile=customServer.cjs --external:next --external:bcrypt --external:@opentelemetry/api --external:deasync --external:cpu-features --loader:.html=text --loader:.scss=text --loader:.node=text",
+    "build:custom-server": "esbuild server.ts --bundle --platform=node --outfile=customServer.cjs --external:next --external:fastify --external:bcrypt --external:@opentelemetry/api --external:deasync --external:cpu-features --loader:.html=text --loader:.scss=text --loader:.node=text",
     "clean": "git clean -xdf .next .turbo node_modules",
     "dev": "pnpm with-env tsx ./server.ts",
     "format": "prettier --check . --ignore-path ../../.gitignore",
@@ -26,7 +26,10 @@
     "@homarr/boards": "workspace:^0.1.0",
     "@homarr/common": "workspace:^0.1.0",
     "@homarr/core": "workspace:^0.1.0",
+    "@homarr/cron-job-api": "workspace:^0.1.0",
     "@homarr/cron-job-status": "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/docker": "workspace:^0.1.0",
@@ -41,7 +44,9 @@
     "@homarr/notifications": "workspace:^0.1.0",
     "@homarr/old-import": "workspace:^0.1.0",
     "@homarr/old-schema": "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/settings": "workspace:^0.1.0",
     "@homarr/spotlight": "workspace:^0.1.0",
@@ -72,6 +77,7 @@
     "clsx": "^2.1.1",
     "dayjs": "^1.11.19",
     "dotenv": "^17.3.1",
+    "fastify": "^5.7.4",
     "flag-icons": "^7.5.0",
     "glob": "^13.0.6",
     "isomorphic-dompurify": "^3.0.0",

apps/nextjs/server.ts

@@ -1,21 +1,223 @@
 // Must be imported first to set up globalThis.AsyncLocalStorage for Next.js
 import "next/dist/server/node-environment-baseline";
 
+// Set up undici HTTP agent override (from tasks app)
+import { setGlobalDispatcher } from "undici";
+import { UndiciHttpAgent } from "@homarr/core/infrastructure/http";
+setGlobalDispatcher(new UndiciHttpAgent());
+
 import { createServer } from "http";
 import path from "path";
 import next from "next";
 import { parse } from "url";
 import { applyWSSHandler } from "@trpc/server/adapters/ws";
 import { WebSocketServer } from "ws";
+import type { FastifyTRPCPluginOptions } from "@trpc/server/adapters/fastify";
+import { fastifyTRPCPlugin } from "@trpc/server/adapters/fastify";
+import fastify from "fastify";
 
 import { appRouter, createTRPCContext } from "@homarr/api/websocket";
 import { getSessionFromToken, sessionTokenCookieName } from "@homarr/auth";
+import { env as authEnv } from "@homarr/auth/env";
 import { parseCookies } from "@homarr/common";
 import { createLogger } from "@homarr/core/infrastructure/logs";
-import { db } from "@homarr/db";
+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, eq, inArray } from "@homarr/db";
+import { cronJobConfigurations, sessions, users } from "@homarr/db/schema";
+import { supportedAuthProviders } from "@homarr/definitions";
+import { updateCheckerRequestHandler } from "@homarr/request-handler/update-checker";
+import { schedule, validate as validateCron } from "node-cron";
 
 const logger = createLogger({ module: "customServer" });
 
+// --- JobManager (from apps/tasks/src/job-manager.ts) ---
+import type { IJobManager } from "@homarr/cron-job-api";
+import type { JobGroupKeys } from "@homarr/cron-jobs";
+import type { Database, InferInsertModel } from "@homarr/db";
+
+class JobManager implements IJobManager {
+  constructor(
+    private db: Database,
+    private jobGroup: typeof jobGroup,
+  ) {}
+
+  public async startAsync(name: JobGroupKeys): Promise<void> {
+    await this.jobGroup.startAsync(name);
+  }
+  public async triggerAsync(name: JobGroupKeys): Promise<void> {
+    await this.jobGroup.runManuallyAsync(name);
+  }
+  public async stopAsync(name: JobGroupKeys): Promise<void> {
+    await this.jobGroup.stopAsync(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`);
+    if (!validateCron(cron)) {
+      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<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 });
+    await this.jobGroup.stopAsync(name);
+    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 });
+    await this.jobGroup.startAsync(name);
+    logger.info("Cron job enabled", { 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),
+    });
+
+    if (existingConfig) {
+      await this.db
+        .update(cronJobConfigurations)
+        .set({ ...configuration, name: undefined })
+        .where(eq(cronJobConfigurations.name, name));
+      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,
+    });
+  }
+
+  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,
+      };
+    });
+  }
+}
+
+// --- On-start functions (from apps/tasks/src/on-start/) ---
+async function cleanupSessionsAsync() {
+  try {
+    const currentAuthProviders = authEnv.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 });
+    }
+  } catch (error) {
+    logger.error(new Error("Failed to clean up sessions", { cause: error }));
+  }
+}
+
+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 }));
+  }
+}
+
+// --- Tasks/Cron Fastify server setup ---
+async function startTasksServer() {
+  const tasksLogger = createLogger({ module: "tasksMain" });
+
+  const tasksServer = fastify({ maxParamLength: 5000 });
+  tasksServer.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 }) {
+        tasksLogger.error(new ErrorWithMetadata("Error in tasks tRPC handler", { path }, { cause: error }));
+      },
+    } satisfies FastifyTRPCPluginOptions<JobRouter>["trpcOptions"],
+  });
+
+  // Run on-start tasks
+  await cleanupSessionsAsync();
+  await invalidateUpdateCheckerCacheAsync();
+
+  // Initialize and start cron jobs
+  await jobGroup.initializeAsync();
+  await jobGroup.startAllAsync();
+
+  try {
+    await tasksServer.listen({ port: CRON_JOB_API_PORT });
+    tasksLogger.info("Tasks web server started successfully", { port: CRON_JOB_API_PORT });
+  } catch (err) {
+    tasksLogger.error(
+      new ErrorWithMetadata("Failed to start tasks web server", { port: CRON_JOB_API_PORT }, { cause: err }),
+    );
+  }
+
+  return tasksServer;
+}
+
+// --- Main server setup ---
 const dev = process.env.NODE_ENV !== "production";
 const hostname = process.env.HOSTNAME || "localhost";
 const port = parseInt(process.env.PORT || "3000", 10);
@@ -25,7 +227,7 @@ const dir = path.resolve(__dirname);
 const app = next({ dev, hostname, port, dir });
 const handle = app.getRequestHandler();
 
-app.prepare().then(() => {
+app.prepare().then(async () => {
   const server = createServer((req, res) => {
     const parsedUrl = parse(req.url!, true);
     handle(req, res, parsedUrl);
@@ -85,15 +287,20 @@ app.prepare().then(() => {
     });
   });
 
+  // Start the integrated tasks/cron server
+  const tasksServer = await startTasksServer();
+
   server.listen(port, () => {
     logger.info(`✅ Custom server ready on http://${hostname}:${port}`);
     logger.info(`✅ WebSocket Server integrated on ws://${hostname}:${port}/websockets`);
+    logger.info(`✅ Tasks/Cron server integrated on port ${CRON_JOB_API_PORT}`);
   });
 
   process.on("SIGTERM", () => {
     logger.info("SIGTERM");
     wssHandler.broadcastReconnectNotification();
     wss.close();
+    tasksServer.close();
     server.close();
   });
 });

apps/tasks/package.json

@@ -12,7 +12,7 @@
   "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",
+    "dev": "echo 'Tasks server is now integrated into the Next.js custom server'",
     "format": "prettier --check . --ignore-path ../../.gitignore",
     "lint": "eslint",
     "typecheck": "tsc --noEmit",

package.json

@@ -26,7 +26,7 @@
     "release": "semantic-release",
     "scripts:update-bug-report-template": "tsx ./scripts/update-bug-report-template.mts",
     "scripts:update-readme-integrations": "tsx ./scripts/update-integration-list.mts",
-    "start": "concurrently \"pnpm with-env node apps/tasks/tasks.cjs\" \"pnpm -F nextjs start\"",
+    "start": "pnpm -F nextjs start",
     "test": "cross-env NODE_ENV=development CI=true vitest run --exclude e2e --coverage.enabled ",
     "test:e2e": "cross-env NODE_ENV=development CI=true vitest e2e",
     "test:ui": "cross-env NODE_ENV=development CI=true vitest --exclude e2e --ui --coverage.enabled",

pnpm-lock.yaml

@@ -139,9 +139,18 @@ importers:
       '@homarr/core':
         specifier: workspace:^0.1.0
         version: link:../../packages/core
+      '@homarr/cron-job-api':
+        specifier: workspace:^0.1.0
+        version: link:../../packages/cron-job-api
       '@homarr/cron-job-status':
         specifier: workspace:^0.1.0
         version: link:../../packages/cron-job-status
+      '@homarr/cron-jobs':
+        specifier: workspace:^0.1.0
+        version: link:../../packages/cron-jobs
+      '@homarr/cron-jobs-core':
+        specifier: workspace:^0.1.0
+        version: link:../../packages/cron-jobs-core
       '@homarr/db':
         specifier: workspace:^0.1.0
         version: link:../../packages/db
@@ -184,9 +193,15 @@ importers:
       '@homarr/old-schema':
         specifier: workspace:^0.1.0
         version: link:../../packages/old-schema
+      '@homarr/ping':
+        specifier: workspace:^0.1.0
+        version: link:../../packages/ping
       '@homarr/redis':
         specifier: workspace:^0.1.0
         version: link:../../packages/redis
+      '@homarr/request-handler':
+        specifier: workspace:^0.1.0
+        version: link:../../packages/request-handler
       '@homarr/server-settings':
         specifier: workspace:^0.1.0
         version: link:../../packages/server-settings
@@ -277,6 +292,9 @@ importers:
       dotenv:
         specifier: ^17.3.1
         version: 17.3.1
+      fastify:
+        specifier: ^5.7.4
+        version: 5.7.4
       flag-icons:
         specifier: ^7.5.0
         version: 7.5.0

scripts/run.sh

@@ -45,17 +45,14 @@ fi
 
 
 
-node apps/tasks/tasks.cjs &
-TASKS_PID=$!
-
-# WebSocket server is now integrated into the custom Next.js server
+# Tasks/cron and WebSocket servers are now integrated into the custom Next.js server
 node apps/nextjs/customServer.cjs &
 NEXTJS_PID=$!
 
 # Function to handle SIGTERM and shut down services
 terminate() {
     echo "Received SIGTERM. Shutting down..."
-    kill -TERM $NGINX_PID $TASKS_PID $NEXTJS_PID 2>/dev/null
+    kill -TERM $NGINX_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