diff --git a/Dockerfile b/Dockerfile index 0edfeaa9d..3ffb327ff 100644 --- a/Dockerfile +++ b/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 diff --git a/apps/nextjs/package.json b/apps/nextjs/package.json index a5d5a5994..03366b748 100644 --- a/apps/nextjs/package.json +++ b/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", diff --git a/apps/nextjs/server.ts b/apps/nextjs/server.ts index 4269e5771..3bea13076 100644 --- a/apps/nextjs/server.ts +++ b/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 { + await this.jobGroup.startAsync(name); + } + public async triggerAsync(name: JobGroupKeys): Promise { + await this.jobGroup.runManuallyAsync(name); + } + public async stopAsync(name: JobGroupKeys): Promise { + await this.jobGroup.stopAsync(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)) { + 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 }); + } + + private async updateConfigurationAsync( + name: JobGroupKeys, + configuration: Omit>, "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["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(); }); }); diff --git a/apps/tasks/package.json b/apps/tasks/package.json index 1a8e4d7c6..03a39c32c 100644 --- a/apps/tasks/package.json +++ b/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", diff --git a/package.json b/package.json index 201c15bbd..4ad6cc08c 100644 --- a/package.json +++ b/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", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7ed0870c8..be2fe1967 100644 --- a/pnpm-lock.yaml +++ b/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 diff --git a/scripts/run.sh b/scripts/run.sh index a3b35ae8b..c92d03543 100644 --- a/scripts/run.sh +++ b/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