feat(tasks): allow management of job intervals and disabling them (#3408)
This commit is contained in:
Meier Lukas
@@ -1,89 +1,216 @@
|
||||
"use client";
|
||||
|
||||
import React from "react";
|
||||
import { ActionIcon, Badge, Card, Group, Stack, Text } from "@mantine/core";
|
||||
import { useListState } from "@mantine/hooks";
|
||||
import { IconPlayerPlay } from "@tabler/icons-react";
|
||||
import React, { useState, useTransition } from "react";
|
||||
import { ActionIcon, Badge, Button, Card, Group, Select, Stack, Text } from "@mantine/core";
|
||||
import { useMap } from "@mantine/hooks";
|
||||
import { IconPlayerPlay, IconPower, IconSettings } from "@tabler/icons-react";
|
||||
|
||||
import type { RouterOutputs } from "@homarr/api";
|
||||
import { clientApi } from "@homarr/api/client";
|
||||
import { useTimeAgo } from "@homarr/common";
|
||||
import { getMantineColor, useTimeAgo } from "@homarr/common";
|
||||
import type { TaskStatus } from "@homarr/cron-job-status";
|
||||
import type { TranslationKeys } from "@homarr/translation";
|
||||
import { useScopedI18n } from "@homarr/translation/client";
|
||||
import { useForm } from "@homarr/form";
|
||||
import { createModal, useModalAction } from "@homarr/modals";
|
||||
import { TranslationFunction } from "@homarr/translation";
|
||||
import { useI18n, useScopedI18n } from "@homarr/translation/client";
|
||||
import { IconPowerOff } from "@homarr/ui/icons";
|
||||
|
||||
interface JobsListProps {
|
||||
initialJobs: RouterOutputs["cronJobs"]["getJobs"];
|
||||
}
|
||||
|
||||
interface JobState {
|
||||
job: JobsListProps["initialJobs"][number];
|
||||
type JobName = RouterOutputs["cronJobs"]["getJobs"][number]["name"];
|
||||
|
||||
export const JobsList = ({ initialJobs }: JobsListProps) => {
|
||||
const [jobs] = clientApi.cronJobs.getJobs.useSuspenseQuery(undefined, {
|
||||
initialData: initialJobs,
|
||||
refetchOnMount: false,
|
||||
});
|
||||
|
||||
const jobStatusMap = useMap<string, TaskStatus | null>(initialJobs.map(({ name }) => [name, null] as const));
|
||||
|
||||
clientApi.cronJobs.subscribeToStatusUpdates.useSubscription(undefined, {
|
||||
onData: (data) => {
|
||||
jobStatusMap.set(data.name, data);
|
||||
},
|
||||
});
|
||||
|
||||
return (
|
||||
<Stack>
|
||||
{jobs.map((job) => {
|
||||
const status = jobStatusMap.get(job.name);
|
||||
|
||||
return <JobCard key={job.name} job={job} status={status ?? null} />;
|
||||
})}
|
||||
</Stack>
|
||||
);
|
||||
};
|
||||
|
||||
const cronExpressions = [
|
||||
{
|
||||
value: "*/5 * * * * *",
|
||||
label: (t) => t("management.page.tool.tasks.interval.seconds", { interval: 5 }),
|
||||
},
|
||||
{
|
||||
value: "*/10 * * * * *",
|
||||
label: (t) => t("management.page.tool.tasks.interval.seconds", { interval: 10 }),
|
||||
},
|
||||
{
|
||||
value: "*/20 * * * * *",
|
||||
label: (t) => t("management.page.tool.tasks.interval.seconds", { interval: 20 }),
|
||||
},
|
||||
{
|
||||
value: "* * * * *",
|
||||
label: (t) => t("management.page.tool.tasks.interval.minutes", { interval: 1 }),
|
||||
},
|
||||
{
|
||||
value: "*/5 * * * *",
|
||||
label: (t) => t("management.page.tool.tasks.interval.minutes", { interval: 5 }),
|
||||
},
|
||||
{
|
||||
value: "*/10 * * * *",
|
||||
label: (t) => t("management.page.tool.tasks.interval.minutes", { interval: 10 }),
|
||||
},
|
||||
{
|
||||
value: "*/15 * * * *",
|
||||
label: (t) => t("management.page.tool.tasks.interval.minutes", { interval: 15 }),
|
||||
},
|
||||
{
|
||||
value: "0 * * * *",
|
||||
label: (t) => t("management.page.tool.tasks.interval.hours", { interval: 1 }),
|
||||
},
|
||||
{
|
||||
value: "0 0 * * */1",
|
||||
label: (t) => t("management.page.tool.tasks.interval.midnight"),
|
||||
},
|
||||
{
|
||||
value: "0 0 * * 1",
|
||||
label: (t) => t("management.page.tool.tasks.interval.weeklyMonday"),
|
||||
},
|
||||
] satisfies { value: string; label: (t: TranslationFunction) => string }[];
|
||||
|
||||
interface JobCardProps {
|
||||
job: RouterOutputs["cronJobs"]["getJobs"][number];
|
||||
status: TaskStatus | null;
|
||||
}
|
||||
|
||||
export const JobsList = ({ initialJobs }: JobsListProps) => {
|
||||
const t = useScopedI18n("management.page.tool.tasks");
|
||||
const [jobs, handlers] = useListState<JobState>(
|
||||
initialJobs.map((job) => ({
|
||||
job,
|
||||
status: null,
|
||||
})),
|
||||
);
|
||||
clientApi.cronJobs.subscribeToStatusUpdates.useSubscription(undefined, {
|
||||
onData: (data) => {
|
||||
const jobByName = jobs.find((job) => job.job.name === data.name);
|
||||
if (!jobByName) {
|
||||
return;
|
||||
}
|
||||
handlers.applyWhere(
|
||||
(job) => job.job.name === data.name,
|
||||
(job) => ({ ...job, status: data }),
|
||||
);
|
||||
},
|
||||
});
|
||||
const { mutateAsync } = clientApi.cronJobs.triggerJob.useMutation();
|
||||
const JobCard = ({ job, status }: JobCardProps) => {
|
||||
const t = useI18n();
|
||||
const tTasks = useScopedI18n("management.page.tool.tasks");
|
||||
const triggerMutation = clientApi.cronJobs.triggerJob.useMutation();
|
||||
const handleJobTrigger = React.useCallback(
|
||||
async (job: JobState) => {
|
||||
if (job.status?.status === "running") {
|
||||
return;
|
||||
}
|
||||
await mutateAsync(job.job.name);
|
||||
async (name: JobName) => {
|
||||
if (status?.status === "running") return;
|
||||
await triggerMutation.mutateAsync(name);
|
||||
},
|
||||
[mutateAsync],
|
||||
[triggerMutation, status],
|
||||
);
|
||||
return (
|
||||
<Stack>
|
||||
{jobs.map((job) => (
|
||||
<Card key={job.job.name} withBorder>
|
||||
<Group justify={"space-between"} gap={"md"}>
|
||||
<Stack gap={0}>
|
||||
<Group>
|
||||
<Text>{t(`job.${job.job.name}.label` as TranslationKeys)}</Text>
|
||||
{job.status?.status === "idle" && <Badge variant="default">{t("status.idle")}</Badge>}
|
||||
{job.status?.status === "running" && <Badge color="green">{t("status.running")}</Badge>}
|
||||
{job.status?.lastExecutionStatus === "error" && <Badge color="red">{t("status.error")}</Badge>}
|
||||
</Group>
|
||||
{job.status && <TimeAgo timestamp={job.status.lastExecutionTimestamp} />}
|
||||
</Stack>
|
||||
|
||||
{!job.job.preventManualExecution && (
|
||||
<ActionIcon
|
||||
onClick={() => handleJobTrigger(job)}
|
||||
disabled={job.status?.status === "running"}
|
||||
variant={"default"}
|
||||
size={"xl"}
|
||||
radius={"xl"}
|
||||
>
|
||||
<IconPlayerPlay stroke={1.5} />
|
||||
</ActionIcon>
|
||||
const { openModal } = useModalAction(TaskConfigurationModal);
|
||||
const [isEnabled, setEnabled] = useState(job.isEnabled);
|
||||
const disableMutation = clientApi.cronJobs.disableJob.useMutation();
|
||||
const enableMutation = clientApi.cronJobs.enableJob.useMutation();
|
||||
|
||||
const [activeStatePending, startActiveTransition] = useTransition();
|
||||
const handleActiveChange = () =>
|
||||
startActiveTransition(async () => {
|
||||
if (isEnabled) {
|
||||
await disableMutation.mutateAsync(job.name, {
|
||||
onSuccess() {
|
||||
setEnabled(false);
|
||||
},
|
||||
});
|
||||
} else {
|
||||
await enableMutation.mutateAsync(job.name, {
|
||||
onSuccess() {
|
||||
setEnabled(true);
|
||||
},
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
return (
|
||||
<Card key={job.name} withBorder>
|
||||
<Group justify={"space-between"} gap={"md"}>
|
||||
<Stack gap={0}>
|
||||
<Group>
|
||||
<Text>{tTasks(`job.${job.name}.label`)}</Text>
|
||||
<StatusBadge isEnabled={isEnabled} status={status} />
|
||||
{status?.lastExecutionStatus === "error" && <Badge color="red">{tTasks("status.error")}</Badge>}
|
||||
</Group>
|
||||
<Group gap="xs">
|
||||
{status && (
|
||||
<>
|
||||
<TimeAgo timestamp={status.lastExecutionTimestamp} />
|
||||
<Text size="sm" c="dimmed">
|
||||
•
|
||||
</Text>
|
||||
<Text size="sm" c="dimmed">
|
||||
{cronExpressions.find((expression) => expression.value === job.cron)?.label(t) ?? job.cron}
|
||||
</Text>
|
||||
</>
|
||||
)}
|
||||
</Group>
|
||||
</Card>
|
||||
))}
|
||||
</Stack>
|
||||
</Stack>
|
||||
|
||||
<Group>
|
||||
{!job.preventManualExecution && (
|
||||
<ActionIcon
|
||||
onClick={() => handleJobTrigger(job.name)}
|
||||
disabled={status?.status === "running"}
|
||||
loading={triggerMutation.isPending}
|
||||
variant="default"
|
||||
size="xl"
|
||||
radius="xl"
|
||||
>
|
||||
<IconPlayerPlay color={getMantineColor("green", 6)} stroke={1.5} />
|
||||
</ActionIcon>
|
||||
)}
|
||||
<ActionIcon onClick={handleActiveChange} loading={activeStatePending} variant="default" size="xl" radius="xl">
|
||||
{isEnabled ? (
|
||||
<IconPower color={getMantineColor("green", 6)} stroke={1.5} />
|
||||
) : (
|
||||
<IconPowerOff color={getMantineColor("gray", 6)} stroke={1.5} />
|
||||
)}
|
||||
</ActionIcon>
|
||||
<ActionIcon
|
||||
onClick={() =>
|
||||
openModal(
|
||||
{ job },
|
||||
{
|
||||
title: tTasks("settings.title", {
|
||||
jobName: tTasks(`job.${job.name}.label`),
|
||||
}),
|
||||
},
|
||||
)
|
||||
}
|
||||
variant={"default"}
|
||||
size={"xl"}
|
||||
radius={"xl"}
|
||||
>
|
||||
<IconSettings stroke={1.5} />
|
||||
</ActionIcon>
|
||||
</Group>
|
||||
</Group>
|
||||
</Card>
|
||||
);
|
||||
};
|
||||
|
||||
interface StatusBadgeProps {
|
||||
isEnabled: boolean;
|
||||
status: TaskStatus | null;
|
||||
}
|
||||
|
||||
const StatusBadge = ({ isEnabled, status }: StatusBadgeProps) => {
|
||||
const t = useScopedI18n("management.page.tool.tasks");
|
||||
if (!isEnabled) return <Badge color="yellow">{t("status.disabled")}</Badge>;
|
||||
|
||||
if (!status) return null;
|
||||
|
||||
if (status.status === "running") return <Badge color="green">{t("status.running")}</Badge>;
|
||||
return <Badge variant="default">{t("status.idle")}</Badge>;
|
||||
};
|
||||
|
||||
const TimeAgo = ({ timestamp }: { timestamp: string }) => {
|
||||
const timeAgo = useTimeAgo(new Date(timestamp));
|
||||
|
||||
@@ -93,3 +220,65 @@ const TimeAgo = ({ timestamp }: { timestamp: string }) => {
|
||||
</Text>
|
||||
);
|
||||
};
|
||||
|
||||
const TaskConfigurationModal = createModal<{
|
||||
job: RouterOutputs["cronJobs"]["getJobs"][number];
|
||||
}>(({ actions, innerProps }) => {
|
||||
const t = useI18n();
|
||||
const form = useForm({
|
||||
initialValues: {
|
||||
cron: innerProps.job.cron,
|
||||
},
|
||||
});
|
||||
const { mutateAsync, isPending } = clientApi.cronJobs.updateJobInterval.useMutation();
|
||||
const utils = clientApi.useUtils();
|
||||
|
||||
return (
|
||||
<form
|
||||
onSubmit={form.onSubmit(async (values) => {
|
||||
utils.cronJobs.getJobs.setData(undefined, (data) =>
|
||||
data?.map((job) =>
|
||||
job.name === innerProps.job.name
|
||||
? {
|
||||
...job,
|
||||
cron: values.cron,
|
||||
}
|
||||
: job,
|
||||
),
|
||||
);
|
||||
await mutateAsync(
|
||||
{
|
||||
name: innerProps.job.name,
|
||||
cron: values.cron,
|
||||
},
|
||||
{
|
||||
onSuccess() {
|
||||
actions.closeModal();
|
||||
},
|
||||
async onSettled() {
|
||||
await utils.cronJobs.getJobs.invalidate();
|
||||
},
|
||||
},
|
||||
);
|
||||
})}
|
||||
>
|
||||
<Stack gap="sm">
|
||||
<Select
|
||||
label={t("management.page.tool.tasks.field.interval.label")}
|
||||
{...form.getInputProps("cron")}
|
||||
data={cronExpressions.map(({ value, label }) => ({ value, label: label(t) }))}
|
||||
/>
|
||||
<Group justify="end">
|
||||
<Button variant="subtle" color="gray" disabled={isPending} onClick={actions.closeModal}>
|
||||
{t("common.action.cancel")}
|
||||
</Button>
|
||||
<Button type="submit" loading={isPending}>
|
||||
{t("common.action.save")}
|
||||
</Button>
|
||||
</Group>
|
||||
</Stack>
|
||||
</form>
|
||||
);
|
||||
}).withOptions({
|
||||
defaultTitle: "",
|
||||
});
|
||||
|
||||
@@ -22,7 +22,7 @@
|
||||
"dependencies": {
|
||||
"@homarr/analytics": "workspace:^0.1.0",
|
||||
"@homarr/common": "workspace:^0.1.0",
|
||||
"@homarr/cron-job-runner": "workspace:^0.1.0",
|
||||
"@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",
|
||||
@@ -37,6 +37,7 @@
|
||||
"@homarr/widgets": "workspace:^0.1.0",
|
||||
"dayjs": "^1.11.13",
|
||||
"dotenv": "^17.0.1",
|
||||
"fastify": "^5.4.0",
|
||||
"superjson": "2.2.2",
|
||||
"undici": "7.11.0"
|
||||
},
|
||||
|
||||
109
apps/tasks/src/job-manager.ts
Normal file
109
apps/tasks/src/job-manager.ts
Normal file
@@ -0,0 +1,109 @@
|
||||
import { schedule, validate as validateCron } from "node-cron";
|
||||
|
||||
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";
|
||||
import { logger } from "@homarr/log";
|
||||
|
||||
export class JobManager implements IJobManager {
|
||||
constructor(
|
||||
private db: Database,
|
||||
private jobGroup: typeof cronJobGroup,
|
||||
) {}
|
||||
|
||||
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="${name}" expression="${cron}"`);
|
||||
const job = this.jobGroup.getJobRegistry().get(name);
|
||||
if (!job) throw new Error(`Job ${name} not found`);
|
||||
if (job.cronExpression === "never") throw new Error(`Job ${name} cannot be updated as it is set to "never"`);
|
||||
if (!validateCron(cron)) {
|
||||
throw new Error(`Invalid cron expression: ${cron}`);
|
||||
}
|
||||
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="${name}" expression="${cron}"`);
|
||||
}
|
||||
public async disableAsync(name: JobGroupKeys): Promise<void> {
|
||||
logger.info(`Disabling cron job name="${name}"`);
|
||||
const job = this.jobGroup.getJobRegistry().get(name);
|
||||
if (!job) throw new Error(`Job ${name} not found`);
|
||||
if (job.cronExpression === "never") throw new Error(`Job ${name} cannot be disabled as it is set to "never"`);
|
||||
|
||||
await this.updateConfigurationAsync(name, { isEnabled: false });
|
||||
await this.jobGroup.stopAsync(name);
|
||||
logger.info(`Cron job disabled name="${name}"`);
|
||||
}
|
||||
public async enableAsync(name: JobGroupKeys): Promise<void> {
|
||||
logger.info(`Enabling cron job name="${name}"`);
|
||||
await this.updateConfigurationAsync(name, { isEnabled: true });
|
||||
await this.jobGroup.startAsync(name);
|
||||
logger.info(`Cron job enabled name="${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="${name}" configuration="${JSON.stringify(configuration)}" exists="${Boolean(existingConfig)}"`,
|
||||
);
|
||||
|
||||
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 name="${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 updated name="${name}" configuration="${JSON.stringify(configuration)}"`);
|
||||
}
|
||||
|
||||
public async getAllAsync(): Promise<
|
||||
{ name: JobGroupKeys; cron: string; preventManualExecution: 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,
|
||||
isEnabled: config?.isEnabled ?? true,
|
||||
};
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -1,10 +1,45 @@
|
||||
// This import has to be the first import in the file so that the agent is overridden before any other modules are imported.
|
||||
import "./undici-log-agent-override";
|
||||
|
||||
import { registerCronJobRunner } from "@homarr/cron-job-runner/register";
|
||||
import type { FastifyTRPCPluginOptions } from "@trpc/server/adapters/fastify";
|
||||
import { fastifyTRPCPlugin } from "@trpc/server/adapters/fastify";
|
||||
import fastify from "fastify";
|
||||
|
||||
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 { logger } from "@homarr/log";
|
||||
|
||||
import { JobManager } from "./job-manager";
|
||||
|
||||
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 Error(`Error in tasks tRPC handler path="${path}"`, { cause: error }));
|
||||
},
|
||||
} satisfies FastifyTRPCPluginOptions<JobRouter>["trpcOptions"],
|
||||
});
|
||||
|
||||
void (async () => {
|
||||
registerCronJobRunner();
|
||||
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 Error(`Failed to start tasks web server port="${CRON_JOB_API_PORT}"`, { cause: err }));
|
||||
process.exit(1);
|
||||
}
|
||||
})();
|
||||
|
||||
Reference in New Issue
Block a user