Add bot schedules (routines), open Activity, add Help and a Desktop computer view

Bots create, list, update and delete schedules (cron_jobs, Bun.cron in UTC); a per-minute cron queues due runs as tasks. Routines, Activity (super admins) and Help get their own rail aside. Help covers familiar problems, starting with signing in to Google through Cookie-Editor over the new Desktop (noVNC) view. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-19 10:52:32 +01:00
parent f24c0d7e82
commit c44048dbab
62 changed files with 1904 additions and 78 deletions
@@ -0,0 +1,108 @@
+import BunSQLite from "@moduletrace/bun-sqlite";
+import type { BUN_SQLITE_COBBLE_CRON_JOBS, BunSQLiteTables } from "@/db/types/db";
+import { AppData } from "@/src/data/app-data";
+import AppError from "@/src/utils/app-error";
+import nextCronRun from "@/src/utils/next-cron-run";
+import grabUserConversation from "../bot-sessions/grab-user-conversation";
+import grabBotComputers from "../computers/grab-bot-computers";
+import grabUserCronJob from "./grab-user-cron-job";
+
+type Params = {
+    user_id: number | string;
+    bot_id: number | string;
+    /**
+     * The workspace it runs in; none for the bot's own chat.
+     */
+    workspace_id?: number | string | null;
+    title: string;
+    instructions: string;
+    /**
+     * A Bun.cron expression, in UTC.
+     */
+    schedule: string;
+    /**
+     * Must be one of the bot's computers there; none runs on its current
+     * one.
+     */
+    computer_id?: number | string | null;
+    max_runs?: number | null;
+};
+
+/**
+ * Gives one of the user's bots a schedule: its brief is queued as a task
+ * at every run (`fireCronJob`), in the bot's conversation there. Returns
+ * the new schedule.
+ */
+export default async function createCronJob({
+    user_id,
+    bot_id,
+    workspace_id,
+    title,
+    instructions,
+    schedule,
+    computer_id,
+    max_runs,
+}: Params) {
+    const conversation = await grabUserConversation({ user_id, bot_id, workspace_id });
+    const bot_id_num = conversation.bot.id;
+
+    const count = Number(
+        (
+            await BunSQLite.sql<{ n: number }>({
+                sql: `SELECT COUNT(*) AS n FROM cron_jobs WHERE bot_id = ?`,
+                values: [bot_id_num],
+            })
+        ).payload?.[0]?.n || 0,
+    );
+
+    if (count >= AppData.CronJobMaxPerBot) {
+        throw new AppError(
+            `${conversation.bot.name || "This bot"} already has ${AppData.CronJobMaxPerBot} schedules. Delete one first.`,
+        );
+    }
+
+    if (computer_id) {
+        const computers = await grabBotComputers({
+            bot_id: bot_id_num,
+            user_id,
+            workspace_id: conversation.workspace_id,
+        });
+
+        if (!computers.some((c) => String(c.id) == String(computer_id))) {
+            throw new AppError(`This bot has no access to that computer.`);
+        }
+    }
+
+    const next_run_at = nextCronRun({ schedule, check_interval: true });
+
+    if (!next_run_at) {
+        throw new AppError(`"${schedule}" never runs.`);
+    }
+
+    const res = await BunSQLite.insert<BUN_SQLITE_COBBLE_CRON_JOBS, (typeof BunSQLiteTables)[number]>({
+        table: "cron_jobs",
+        data: [
+            {
+                user_id: Number(user_id),
+                bot_id: bot_id_num,
+                workspace_id: conversation.workspace_id || undefined,
+                title,
+                instructions,
+                schedule: schedule.trim(),
+                computer_id: computer_id ? Number(computer_id) : undefined,
+                enabled: 1,
+                max_runs: max_runs || undefined,
+                run_count: 0,
+                next_run_at,
+            },
+        ],
+    });
+
+    const cron_job_id = res.singleRes?.id || res.postInsertReturn?.insertId;
+
+    if (!res.success || !cron_job_id) {
+        throw new Error(`Couldn't save the schedule: ${res.msg}`);
+    }
+
+    return await grabUserCronJob({ user_id, cron_job_id });
+}
@@ -0,0 +1,29 @@
+import BunSQLite from "@moduletrace/bun-sqlite";
+import grabUserCronJob from "./grab-user-cron-job";
+
+type Params = {
+    user_id: number | string;
+    cron_job_id: number | string;
+    /**
+     * Only if it's this bot's (a bot managing its own schedules).
+     */
+    bot_id?: number | string;
+};
+
+/**
+ * Deletes one of the user's schedules. A run already queued carries on.
+ */
+export default async function deleteCronJob({ user_id, cron_job_id, bot_id }: Params) {
+    const cron_job = await grabUserCronJob({ user_id, cron_job_id, bot_id });
+
+    const res = await BunSQLite.sql({
+        sql: `DELETE FROM cron_jobs WHERE id = ?`,
+        values: [cron_job.id!],
+    });
+
+    if (!res.success) {
+        throw new Error(`Couldn't delete the schedule: ${res.error}`);
+    }
+
+    return cron_job;
+}
@@ -0,0 +1,145 @@
+import BunSQLite from "@moduletrace/bun-sqlite";
+import type { BUN_SQLITE_COBBLE_CRON_JOBS } from "@/db/types/db";
+import wakeBotsQueue from "@/src/bots-queue/wake-bots-queue";
+import logEvent, { logError } from "@/src/functions/backend/logger";
+import nextCronRun from "@/src/utils/next-cron-run";
+import safeErrorMessage from "@/src/utils/safe-error-message";
+import ensureBotSession from "../bot-sessions/ensure-bot-session";
+import enqueueBotJob from "../bots-queue/enqueue-bot-job";
+import grabBotComputers from "../computers/grab-bot-computers";
+
+type Params = {
+    cron_job: BUN_SQLITE_COBBLE_CRON_JOBS;
+    now?: number;
+};
+
+/**
+ * Runs one due schedule: moves it to its next run (claimed atomically, so
+ * a run is never queued twice), then queues its brief as a task in the
+ * bot's active session there, on its computer or the bot's current one.
+ * A run is skipped while the last one is still queued or working. It
+ * turns itself off once it has no next run or has done its `max_runs`.
+ * Why a run couldn't be queued is kept in `last_error`.
+ */
+export default async function fireCronJob({ cron_job, now = Date.now() }: Params) {
+    const id = cron_job.id;
+    const schedule = cron_job.schedule || "";
+
+    if (!id || !cron_job.next_run_at) return;
+
+    let next_run_at: number | null = null;
+
+    try {
+        next_run_at = nextCronRun({ schedule, from: now });
+    } catch {}
+
+    const claimed = await BunSQLite.sql<{ id: number }>({
+        sql: `UPDATE cron_jobs SET next_run_at = ?, enabled = ?, updated_at = ?
+            WHERE id = ? AND enabled = 1 AND next_run_at = ? RETURNING id`,
+        values: [next_run_at, next_run_at ? 1 : 0, now, id, cron_job.next_run_at],
+    });
+
+    if (!claimed.payload?.[0]) return;
+
+    try {
+        if (cron_job.last_job_id && (await isJobOpen({ job_id: cron_job.last_job_id }))) {
+            await saveRun({ id, last_error: `Skipped a run: the last one was still going.` });
+            return;
+        }
+
+        const session = await ensureBotSession({
+            bot_id: cron_job.bot_id!,
+            workspace_id: cron_job.workspace_id || null,
+        });
+
+        const computer_id =
+            cron_job.computer_id ||
+            (
+                await grabBotComputers({
+                    bot_id: cron_job.bot_id!,
+                    user_id: cron_job.user_id!,
+                    workspace_id: cron_job.workspace_id || null,
+                })
+            )[0]?.id;
+
+        const job_id = await enqueueBotJob({
+            user_id: cron_job.user_id!,
+            bot_id: cron_job.bot_id!,
+            session_id: session.id!,
+            title: cron_job.title || "Scheduled task",
+            instructions: `${cron_job.instructions || ""}\n\nThis is a scheduled run of your schedule #${id} (\`${schedule}\`, UTC). Do the work; don't create another schedule for it.`,
+            params: { schedule_id: id, schedule, run: Number(cron_job.run_count || 0) + 1 },
+            computer_id: computer_id || undefined,
+        });
+
+        const run_count = Number(cron_job.run_count || 0) + 1;
+        const done = Boolean(cron_job.max_runs && run_count >= cron_job.max_runs);
+
+        await saveRun({ id, last_run_at: now, last_job_id: Number(job_id), run_count, turn_off: done });
+
+        wakeBotsQueue();
+
+        await logEvent({
+            title: `Schedule: ${cron_job.title}`,
+            category: "cron",
+            user_id: cron_job.user_id || undefined,
+            plain_text: `Queued schedule #${id} "${cron_job.title}" (${schedule}) as job #${job_id}.`,
+            metadata: { cron_job_id: id, job_id, bot_id: cron_job.bot_id },
+        });
+    } catch (error) {
+        await saveRun({ id, last_error: safeErrorMessage({ error, fallback: `Couldn't queue the run.` }) });
+
+        await logError({
+            title: `Schedule #${id} couldn't run`,
+            category: "cron",
+            error,
+            user_id: cron_job.user_id || undefined,
+            metadata: { cron_job_id: id },
+        }).catch(() => {});
+    }
+}
+
+async function isJobOpen({ job_id }: { job_id: number }) {
+    const res = await BunSQLite.sql({
+        sql: `SELECT id FROM bots_queue WHERE id = ? AND status IN ('pending', 'processing') LIMIT 1`,
+        values: [job_id],
+    });
+
+    return Boolean(res.payload?.[0]);
+}
+
+async function saveRun({
+    id,
+    last_run_at,
+    last_job_id,
+    run_count,
+    turn_off,
+    last_error = null,
+}: {
+    id: number;
+    last_run_at?: number;
+    last_job_id?: number;
+    run_count?: number;
+    turn_off?: boolean;
+    last_error?: string | null;
+}) {
+    await BunSQLite.sql({
+        sql: `UPDATE cron_jobs SET
+                last_run_at = COALESCE(?, last_run_at),
+                last_job_id = COALESCE(?, last_job_id),
+                run_count = COALESCE(?, run_count),
+                enabled = CASE WHEN ? THEN 0 ELSE enabled END,
+                last_error = ?,
+                updated_at = ?
+            WHERE id = ?`,
+        values: [
+            last_run_at ?? null,
+            last_job_id ?? null,
+            run_count ?? null,
+            turn_off ? 1 : 0,
+            last_error,
+            Date.now(),
+            id,
+        ],
+    });
+}
@@ -0,0 +1,46 @@
+import BunSQLite from "@moduletrace/bun-sqlite";
+import type { CronJobWithNames } from "@/src/types";
+
+type Params = {
+    user_id: number | string;
+    /**
+     * Only this bot's.
+     */
+    bot_id?: number | string;
+    /**
+     * Only this one.
+     */
+    cron_job_id?: number | string;
+};
+
+/**
+ * The user's schedules with their bot and workspace names, the next to
+ * run first and paused ones last.
+ */
+export default async function grabCronJobs({ user_id, bot_id, cron_job_id }: Params): Promise<CronJobWithNames[]> {
+    const where = [`c.user_id = ?`];
+    const values: (number | string)[] = [user_id];
+
+    if (bot_id) {
+        where.push(`c.bot_id = ?`);
+        values.push(bot_id);
+    }
+
+    if (cron_job_id) {
+        where.push(`c.id = ?`);
+        values.push(cron_job_id);
+    }
+
+    const res = await BunSQLite.sql<CronJobWithNames>({
+        sql: `SELECT c.*, b.name AS bot_name, b.kind AS bot_kind, b.hue AS bot_hue, b.face AS bot_face,
+                b.image_media_id AS bot_image_media_id, w.name AS workspace_name
+            FROM cron_jobs c
+            JOIN bots b ON b.id = c.bot_id
+            LEFT JOIN workspaces w ON w.id = c.workspace_id
+            WHERE ${where.join(" AND ")}
+            ORDER BY c.enabled DESC, c.next_run_at IS NULL, c.next_run_at ASC, c.id ASC`,
+        values,
+    });
+
+    return res.payload || [];
+}
@@ -0,0 +1,24 @@
+import AppError from "@/src/utils/app-error";
+import grabCronJobs from "./grab-cron-jobs";
+
+type Params = {
+    user_id: number | string;
+    cron_job_id: number | string;
+    /**
+     * Only if it's this bot's (a bot managing its own schedules).
+     */
+    bot_id?: number | string;
+};
+
+/**
+ * One of the user's schedules. Throws when it isn't theirs (or the bot's).
+ */
+export default async function grabUserCronJob({ user_id, cron_job_id, bot_id }: Params) {
+    const [cron_job] = await grabCronJobs({ user_id, bot_id, cron_job_id });
+
+    if (!cron_job?.id) {
+        throw new AppError(`Schedule #${cron_job_id} not found.`);
+    }
+
+    return cron_job;
+}
@@ -0,0 +1,83 @@
+import BunSQLite from "@moduletrace/bun-sqlite";
+import AppError from "@/src/utils/app-error";
+import nextCronRun from "@/src/utils/next-cron-run";
+import grabUserCronJob from "./grab-user-cron-job";
+
+type Params = {
+    user_id: number | string;
+    cron_job_id: number | string;
+    /**
+     * Only if it's this bot's (a bot managing its own schedules).
+     */
+    bot_id?: number | string;
+    /**
+     * Only the fields given are changed. `max_runs: null` runs it until
+     * it's stopped.
+     */
+    data: {
+        title?: string;
+        instructions?: string;
+        schedule?: string;
+        enabled?: boolean;
+        max_runs?: number | null;
+    };
+};
+
+/**
+ * Changes one of the user's schedules. A new schedule, or turning it back
+ * on, counts the next run from now; turning a finished one back on (all
+ * its `max_runs` done) starts its count over. Returns the schedule.
+ */
+export default async function updateCronJob({ user_id, cron_job_id, bot_id, data }: Params) {
+    const cron_job = await grabUserCronJob({ user_id, cron_job_id, bot_id });
+
+    const schedule = data.schedule?.trim() || cron_job.schedule || "";
+    const schedule_changed = Boolean(data.schedule?.trim()) && schedule != cron_job.schedule;
+    const enabled = data.enabled ?? Boolean(cron_job.enabled);
+    const turned_on = enabled && !cron_job.enabled;
+    const max_runs = data.max_runs === undefined ? cron_job.max_runs || null : data.max_runs;
+
+    const sets: string[] = [];
+    const values: (string | number | null)[] = [];
+
+    const set = (field: string, value: string | number | null) => {
+        sets.push(`${field} = ?`);
+        values.push(value);
+    };
+
+    if (data.title !== undefined) set("title", data.title);
+    if (data.instructions !== undefined) set("instructions", data.instructions);
+    if (data.max_runs !== undefined) set("max_runs", data.max_runs);
+    if (data.enabled !== undefined) set("enabled", enabled ? 1 : 0);
+
+    if (schedule_changed || turned_on) {
+        const next_run_at = nextCronRun({ schedule, check_interval: schedule_changed });
+
+        if (!next_run_at) {
+            throw new AppError(`"${schedule}" never runs.`);
+        }
+
+        set("schedule", schedule);
+        set("next_run_at", next_run_at);
+        set("last_error", null);
+    }
+
+    if (turned_on && max_runs && Number(cron_job.run_count || 0) >= max_runs) {
+        set("run_count", 0);
+    }
+
+    if (!sets.length) return cron_job;
+
+    set("updated_at", Date.now());
+
+    const res = await BunSQLite.sql({
+        sql: `UPDATE cron_jobs SET ${sets.join(", ")} WHERE id = ?`,
+        values: [...values, cron_job.id!],
+    });
+
+    if (!res.success) {
+        throw new Error(`Couldn't update the schedule: ${res.error}`);
+    }
+
+    return await grabUserCronJob({ user_id, cron_job_id: cron_job.id! });
+}