feat: expand scheduling, notifications, and queue runtime
- add scheduled task plumbing across backend core, CLI, and frontend surfaces - add frontend notification UI and keep service pages using the repaired shared stylesheet - refactor code queue runtime and update baidu netdisk/service integration docs
This commit is contained in:
Codex
@@ -1,4 +1,5 @@
|
||||
FROM oven/bun:1-alpine
|
||||
RUN apk add --no-cache postgresql16-client tar gzip
|
||||
WORKDIR /app/src/components/backend-core
|
||||
COPY src/components/backend-core/package.json ./package.json
|
||||
RUN bun install --production
|
||||
|
||||
@@ -1,4 +1,8 @@
|
||||
import type { Server, ServerWebSocket } from "bun";
|
||||
import { createHash } from "node:crypto";
|
||||
import { createReadStream } from "node:fs";
|
||||
import { mkdir, rm, stat } from "node:fs/promises";
|
||||
import { basename, dirname, relative, resolve, posix as pathPosix } from "node:path";
|
||||
import postgres from "postgres";
|
||||
import { createHourlyJsonlWriter, logRetentionBytesForService } from "../../shared/src/rotating-jsonl";
|
||||
import {
|
||||
@@ -33,6 +37,8 @@ interface RuntimeConfig {
|
||||
taskPendingTimeoutMs: number;
|
||||
databaseVolumeName: string;
|
||||
databaseVolumeSize: string;
|
||||
pgdataBackupStagingDir: string;
|
||||
baiduNetdiskInternalUrl: string;
|
||||
microservices: MicroserviceConfig[];
|
||||
logFile: string;
|
||||
}
|
||||
@@ -112,6 +118,63 @@ interface RawTaskRow {
|
||||
updated_at: Date | string;
|
||||
}
|
||||
|
||||
interface ScheduledTaskRow {
|
||||
id: string;
|
||||
name: string;
|
||||
description: string;
|
||||
enabled: boolean;
|
||||
schedule_json: JsonValue;
|
||||
action_json: JsonValue;
|
||||
concurrency_policy: string;
|
||||
next_run_at: Date | string | null;
|
||||
last_run_at: Date | string | null;
|
||||
last_run_id: string | null;
|
||||
created_at: Date | string;
|
||||
updated_at: Date | string;
|
||||
}
|
||||
|
||||
interface ScheduledTaskRunRow {
|
||||
id: string;
|
||||
schedule_id: string;
|
||||
trigger_type: string;
|
||||
status: string;
|
||||
task_id: string | null;
|
||||
result: JsonValue | null;
|
||||
error: string | null;
|
||||
started_at: Date | string | null;
|
||||
finished_at: Date | string | null;
|
||||
duration_ms: number | string | null;
|
||||
created_at: Date | string;
|
||||
updated_at: Date | string;
|
||||
}
|
||||
|
||||
interface ScheduleSpec {
|
||||
type: "daily" | "interval";
|
||||
timeOfDay?: string;
|
||||
timezone?: string;
|
||||
everySeconds?: number;
|
||||
}
|
||||
|
||||
interface DispatchScheduleAction {
|
||||
type: "dispatch";
|
||||
providerId: string;
|
||||
command: CoreDispatchMessage["command"];
|
||||
payload: Record<string, JsonValue>;
|
||||
timeoutMs?: number;
|
||||
}
|
||||
|
||||
interface PgdataBackupScheduleAction {
|
||||
type: "pgdata_backup";
|
||||
volumeName: string;
|
||||
remoteBaseDir: string;
|
||||
stagingSubdir: string;
|
||||
baiduBaseUrl: string;
|
||||
timeoutMs: number;
|
||||
cleanupLocal: boolean;
|
||||
}
|
||||
|
||||
type ScheduleAction = DispatchScheduleAction | PgdataBackupScheduleAction;
|
||||
|
||||
type TaskTerminalWaiter = (task: RawTaskRow | null) => void;
|
||||
|
||||
interface MicroserviceProxyCacheEntry {
|
||||
@@ -141,6 +204,7 @@ const maxPerformanceSamples = 3000;
|
||||
const taskTerminalWaiters = new Map<string, Set<TaskTerminalWaiter>>();
|
||||
const microserviceProxyCache = new Map<string, MicroserviceProxyCacheEntry>();
|
||||
const microserviceProxyRefreshes = new Map<string, Promise<void>>();
|
||||
const activeScheduledRuns = new Set<string>();
|
||||
let lastTaskSweepAt = 0;
|
||||
let taskSweepInFlight: Promise<void> | null = null;
|
||||
const microserviceProxyMaxBodyTextLength = 8 * 1024 * 1024;
|
||||
@@ -280,6 +344,8 @@ function readConfig(): RuntimeConfig {
|
||||
taskPendingTimeoutMs: readOptionalNumberEnv("TASK_PENDING_TIMEOUT_MS", 10 * 60 * 1000),
|
||||
databaseVolumeName: requiredEnv("DATABASE_VOLUME_NAME"),
|
||||
databaseVolumeSize: requiredEnv("DATABASE_VOLUME_SIZE"),
|
||||
pgdataBackupStagingDir: process.env.PGDATA_BACKUP_STAGING_DIR || "/data/baidu-netdisk-staging",
|
||||
baiduNetdiskInternalUrl: process.env.BAIDU_NETDISK_INTERNAL_URL || "http://baidu-netdisk:4244",
|
||||
microservices: readMicroservicesEnv(),
|
||||
logFile: requiredEnv("LOG_FILE"),
|
||||
};
|
||||
@@ -342,6 +408,7 @@ function classifyRequestComponent(pathname: string): string {
|
||||
if (pathname.startsWith("/api/nodes/")) return "node_metrics_api";
|
||||
if (pathname === "/api/nodes") return "node_inventory_api";
|
||||
if (pathname.startsWith("/api/tasks")) return "scheduler_api";
|
||||
if (pathname.startsWith("/api/schedules")) return "scheduler_api";
|
||||
if (pathname.startsWith("/api/events")) return "event_api";
|
||||
if (pathname.startsWith("/api/performance")) return "performance_api";
|
||||
if (pathname.startsWith("/api/")) return "core_api";
|
||||
@@ -556,6 +623,38 @@ async function initDatabase(client: SqlClient): Promise<void> {
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)
|
||||
`;
|
||||
await client`
|
||||
CREATE TABLE IF NOT EXISTS unidesk_scheduled_tasks (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
description TEXT NOT NULL DEFAULT '',
|
||||
enabled BOOLEAN NOT NULL DEFAULT true,
|
||||
schedule_json JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
action_json JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
concurrency_policy TEXT NOT NULL DEFAULT 'skip',
|
||||
next_run_at TIMESTAMPTZ,
|
||||
last_run_at TIMESTAMPTZ,
|
||||
last_run_id TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)
|
||||
`;
|
||||
await client`
|
||||
CREATE TABLE IF NOT EXISTS unidesk_scheduled_task_runs (
|
||||
id TEXT PRIMARY KEY,
|
||||
schedule_id TEXT NOT NULL REFERENCES unidesk_scheduled_tasks(id) ON DELETE CASCADE,
|
||||
trigger_type TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
task_id TEXT,
|
||||
result JSONB,
|
||||
error TEXT,
|
||||
started_at TIMESTAMPTZ,
|
||||
finished_at TIMESTAMPTZ,
|
||||
duration_ms BIGINT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)
|
||||
`;
|
||||
await client`
|
||||
CREATE TABLE IF NOT EXISTS unidesk_node_docker_status (
|
||||
provider_id TEXT PRIMARY KEY,
|
||||
@@ -586,6 +685,9 @@ async function initDatabase(client: SqlClient): Promise<void> {
|
||||
`;
|
||||
await client`CREATE INDEX IF NOT EXISTS idx_unidesk_tasks_updated_at ON unidesk_tasks(updated_at DESC)`;
|
||||
await client`CREATE INDEX IF NOT EXISTS idx_unidesk_tasks_status_updated_at ON unidesk_tasks(status, updated_at DESC)`;
|
||||
await client`CREATE INDEX IF NOT EXISTS idx_unidesk_scheduled_tasks_next_run ON unidesk_scheduled_tasks(enabled, next_run_at)`;
|
||||
await client`CREATE INDEX IF NOT EXISTS idx_unidesk_scheduled_task_runs_schedule_updated ON unidesk_scheduled_task_runs(schedule_id, updated_at DESC)`;
|
||||
await client`CREATE INDEX IF NOT EXISTS idx_unidesk_scheduled_task_runs_status_updated ON unidesk_scheduled_task_runs(status, updated_at DESC)`;
|
||||
await client`CREATE INDEX IF NOT EXISTS idx_unidesk_node_system_status_updated_at ON unidesk_node_system_status(updated_at DESC)`;
|
||||
await client`CREATE INDEX IF NOT EXISTS idx_unidesk_node_metric_samples_provider_time ON unidesk_node_metric_samples(provider_id, collected_at DESC)`;
|
||||
dbReady = true;
|
||||
@@ -1501,6 +1603,720 @@ async function waitForTaskTerminal(taskId: string, timeoutMs: number): Promise<R
|
||||
});
|
||||
}
|
||||
|
||||
function rowIso(value: Date | string | null | undefined): string | null {
|
||||
if (value === null || value === undefined) return null;
|
||||
return value instanceof Date ? value.toISOString() : String(value);
|
||||
}
|
||||
|
||||
function isJsonRecord(value: unknown): value is Record<string, JsonValue> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function scheduleRunView(row: ScheduledTaskRunRow): JsonValue {
|
||||
return {
|
||||
id: row.id,
|
||||
scheduleId: row.schedule_id,
|
||||
trigger: row.trigger_type,
|
||||
status: row.status,
|
||||
taskId: row.task_id,
|
||||
result: compactJson(row.result ?? null),
|
||||
error: row.error ?? "",
|
||||
startedAt: rowIso(row.started_at),
|
||||
finishedAt: rowIso(row.finished_at),
|
||||
durationMs: row.duration_ms === null ? null : Number(row.duration_ms),
|
||||
createdAt: rowIso(row.created_at),
|
||||
updatedAt: rowIso(row.updated_at),
|
||||
};
|
||||
}
|
||||
|
||||
function scheduledTaskView(row: ScheduledTaskRow, runs: ScheduledTaskRunRow[] = []): JsonValue {
|
||||
return {
|
||||
id: row.id,
|
||||
name: row.name,
|
||||
description: row.description,
|
||||
enabled: row.enabled,
|
||||
schedule: row.schedule_json,
|
||||
action: row.action_json,
|
||||
concurrencyPolicy: row.concurrency_policy,
|
||||
nextRunAt: rowIso(row.next_run_at),
|
||||
lastRunAt: rowIso(row.last_run_at),
|
||||
lastRunId: row.last_run_id,
|
||||
createdAt: rowIso(row.created_at),
|
||||
updatedAt: rowIso(row.updated_at),
|
||||
recentRuns: runs.map(scheduleRunView),
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeScheduleId(value: unknown): string {
|
||||
const raw = typeof value === "string" && value.trim().length > 0
|
||||
? value.trim()
|
||||
: `schedule_${Date.now()}_${Math.random().toString(16).slice(2, 8)}`;
|
||||
if (!/^[A-Za-z0-9_.:-]{1,120}$/u.test(raw)) {
|
||||
throw new Error("schedule id must be 1-120 chars using letters, numbers, _, ., :, or -");
|
||||
}
|
||||
return raw;
|
||||
}
|
||||
|
||||
function normalizeTimeOfDay(value: unknown): string {
|
||||
const raw = typeof value === "string" && value.trim().length > 0 ? value.trim() : "03:00";
|
||||
const match = raw.match(/^([01]\d|2[0-3]):([0-5]\d)$/u);
|
||||
if (match === null) throw new Error("daily schedule timeOfDay must use HH:MM in UTC");
|
||||
return `${match[1]}:${match[2]}`;
|
||||
}
|
||||
|
||||
function numberInRange(value: unknown, fallback: number, min: number, max: number): number {
|
||||
const parsed = typeof value === "number" ? value : typeof value === "string" ? Number(value) : fallback;
|
||||
if (!Number.isFinite(parsed)) return fallback;
|
||||
return Math.max(min, Math.min(max, Math.floor(parsed)));
|
||||
}
|
||||
|
||||
function normalizeScheduleSpec(value: unknown): ScheduleSpec {
|
||||
const record = isJsonRecord(value) ? value : {};
|
||||
if (record.type === "interval") {
|
||||
return {
|
||||
type: "interval",
|
||||
everySeconds: numberInRange(record.everySeconds, 3600, 60, 366 * 24 * 3600),
|
||||
};
|
||||
}
|
||||
return {
|
||||
type: "daily",
|
||||
timeOfDay: normalizeTimeOfDay(record.timeOfDay),
|
||||
timezone: typeof record.timezone === "string" && record.timezone.length > 0 ? record.timezone : "Etc/UTC",
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeScheduleAction(value: unknown): ScheduleAction {
|
||||
if (!isJsonRecord(value)) throw new Error("scheduled task action must be an object");
|
||||
if (value.type === "dispatch") {
|
||||
const providerId = typeof value.providerId === "string" ? value.providerId.trim() : "";
|
||||
if (providerId.length === 0) throw new Error("dispatch action providerId is required");
|
||||
if (!isProviderDispatchCommand(value.command)) throw new Error("dispatch action command is invalid");
|
||||
const payload = isJsonRecord(value.payload) ? value.payload : {};
|
||||
const timeoutMs = value.timeoutMs === undefined ? undefined : numberInRange(value.timeoutMs, config.taskPendingTimeoutMs, 1_000, 24 * 3600_000);
|
||||
return { type: "dispatch", providerId, command: value.command, payload, ...(timeoutMs === undefined ? {} : { timeoutMs }) };
|
||||
}
|
||||
if (value.type === "pgdata_backup") {
|
||||
return {
|
||||
type: "pgdata_backup",
|
||||
volumeName: typeof value.volumeName === "string" && value.volumeName.length > 0 ? value.volumeName : config.databaseVolumeName,
|
||||
remoteBaseDir: normalizeRemoteDir(typeof value.remoteBaseDir === "string" ? value.remoteBaseDir : "/SERVER_DATA/UNIDESK_PG_DATA"),
|
||||
stagingSubdir: normalizeRelativeStagingPath(typeof value.stagingSubdir === "string" ? value.stagingSubdir : "server-data/unidesk-pg-data").relativePath,
|
||||
baiduBaseUrl: typeof value.baiduBaseUrl === "string" && value.baiduBaseUrl.length > 0 ? value.baiduBaseUrl : config.baiduNetdiskInternalUrl,
|
||||
timeoutMs: numberInRange(value.timeoutMs, 60 * 60_000, 60_000, 24 * 3600_000),
|
||||
cleanupLocal: value.cleanupLocal !== false,
|
||||
};
|
||||
}
|
||||
throw new Error("scheduled task action.type must be dispatch or pgdata_backup");
|
||||
}
|
||||
|
||||
function computeNextRunAt(schedule: ScheduleSpec, after = new Date()): Date {
|
||||
if (schedule.type === "interval") {
|
||||
return new Date(after.getTime() + Math.max(60, schedule.everySeconds ?? 3600) * 1000);
|
||||
}
|
||||
const [hourText = "03", minuteText = "00"] = (schedule.timeOfDay ?? "03:00").split(":");
|
||||
const candidate = new Date(Date.UTC(after.getUTCFullYear(), after.getUTCMonth(), after.getUTCDate(), Number(hourText), Number(minuteText), 0, 0));
|
||||
if (candidate.getTime() <= after.getTime()) candidate.setUTCDate(candidate.getUTCDate() + 1);
|
||||
return candidate;
|
||||
}
|
||||
|
||||
async function getScheduledTasks(limit: number): Promise<JsonValue[]> {
|
||||
const rows = await sql<ScheduledTaskRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_tasks
|
||||
ORDER BY enabled DESC, next_run_at ASC NULLS LAST, updated_at DESC
|
||||
LIMIT ${limit}
|
||||
`;
|
||||
const runRows = await sql<ScheduledTaskRunRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_task_runs
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT ${Math.max(20, limit * 5)}
|
||||
`;
|
||||
const runsBySchedule = new Map<string, ScheduledTaskRunRow[]>();
|
||||
for (const run of runRows) {
|
||||
const list = runsBySchedule.get(run.schedule_id) ?? [];
|
||||
if (list.length < 5) list.push(run);
|
||||
runsBySchedule.set(run.schedule_id, list);
|
||||
}
|
||||
return rows.map((row) => scheduledTaskView(row, runsBySchedule.get(row.id) ?? []));
|
||||
}
|
||||
|
||||
async function getScheduledTaskRuns(scheduleId: string | null, limit: number): Promise<JsonValue[]> {
|
||||
const rows = scheduleId === null
|
||||
? await sql<ScheduledTaskRunRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_task_runs
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT ${limit}
|
||||
`
|
||||
: await sql<ScheduledTaskRunRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_task_runs
|
||||
WHERE schedule_id = ${scheduleId}
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT ${limit}
|
||||
`;
|
||||
return rows.map(scheduleRunView);
|
||||
}
|
||||
|
||||
async function getScheduledTaskRow(scheduleId: string): Promise<ScheduledTaskRow | null> {
|
||||
const rows = await sql<ScheduledTaskRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_tasks
|
||||
WHERE id = ${scheduleId}
|
||||
LIMIT 1
|
||||
`;
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
async function upsertScheduledTask(req: Request, scheduleIdFromPath: string | null): Promise<Response> {
|
||||
const body = await req.json().catch(() => ({})) as Record<string, unknown>;
|
||||
const id = normalizeScheduleId(scheduleIdFromPath ?? body.id);
|
||||
const schedule = normalizeScheduleSpec(body.schedule);
|
||||
const action = normalizeScheduleAction(body.action);
|
||||
const name = typeof body.name === "string" && body.name.trim().length > 0 ? body.name.trim() : id;
|
||||
const description = typeof body.description === "string" ? body.description : "";
|
||||
const enabled = typeof body.enabled === "boolean" ? body.enabled : true;
|
||||
const concurrencyPolicy = body.concurrencyPolicy === "parallel" ? "parallel" : "skip";
|
||||
const existing = await getScheduledTaskRow(id);
|
||||
const nextRunAt = existing?.next_run_at && JSON.stringify(existing.schedule_json) === JSON.stringify(schedule)
|
||||
? rowIso(existing.next_run_at)
|
||||
: computeNextRunAt(schedule).toISOString();
|
||||
const scheduleJson = schedule as unknown as JsonValue;
|
||||
const actionJson = action as unknown as JsonValue;
|
||||
const rows = await sql<ScheduledTaskRow[]>`
|
||||
INSERT INTO unidesk_scheduled_tasks (id, name, description, enabled, schedule_json, action_json, concurrency_policy, next_run_at, updated_at)
|
||||
VALUES (${id}, ${name}, ${description}, ${enabled}, ${sql.json(scheduleJson)}, ${sql.json(actionJson)}, ${concurrencyPolicy}, ${nextRunAt}, now())
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
name = EXCLUDED.name,
|
||||
description = EXCLUDED.description,
|
||||
enabled = EXCLUDED.enabled,
|
||||
schedule_json = EXCLUDED.schedule_json,
|
||||
action_json = EXCLUDED.action_json,
|
||||
concurrency_policy = EXCLUDED.concurrency_policy,
|
||||
next_run_at = EXCLUDED.next_run_at,
|
||||
updated_at = now()
|
||||
RETURNING *
|
||||
`;
|
||||
await recordEvent(existing === null ? "scheduled_task_created" : "scheduled_task_updated", "scheduler", { scheduleId: id, name, enabled, schedule: scheduleJson, action: compactJson(action) });
|
||||
return jsonResponse({ ok: true, schedule: scheduledTaskView(rows[0]) }, existing === null ? 201 : 200);
|
||||
}
|
||||
|
||||
async function patchScheduledTask(scheduleId: string, req: Request): Promise<Response> {
|
||||
const current = await getScheduledTaskRow(scheduleId);
|
||||
if (current === null) return jsonResponse({ ok: false, error: `scheduled task not found: ${scheduleId}` }, 404);
|
||||
const body = await req.json().catch(() => ({})) as Record<string, unknown>;
|
||||
const schedule = body.schedule === undefined ? normalizeScheduleSpec(current.schedule_json) : normalizeScheduleSpec(body.schedule);
|
||||
const action = body.action === undefined ? normalizeScheduleAction(current.action_json) : normalizeScheduleAction(body.action);
|
||||
const name = typeof body.name === "string" && body.name.trim().length > 0 ? body.name.trim() : current.name;
|
||||
const description = typeof body.description === "string" ? body.description : current.description;
|
||||
const enabled = typeof body.enabled === "boolean" ? body.enabled : current.enabled;
|
||||
const concurrencyPolicy = body.concurrencyPolicy === "parallel" || body.concurrencyPolicy === "skip" ? body.concurrencyPolicy : current.concurrency_policy;
|
||||
const scheduleChanged = body.schedule !== undefined && JSON.stringify(schedule) !== JSON.stringify(current.schedule_json);
|
||||
const nextRunAt = scheduleChanged || current.next_run_at === null ? computeNextRunAt(schedule).toISOString() : rowIso(current.next_run_at);
|
||||
const scheduleJson = schedule as unknown as JsonValue;
|
||||
const actionJson = action as unknown as JsonValue;
|
||||
const rows = await sql<ScheduledTaskRow[]>`
|
||||
UPDATE unidesk_scheduled_tasks
|
||||
SET name = ${name},
|
||||
description = ${description},
|
||||
enabled = ${enabled},
|
||||
schedule_json = ${sql.json(scheduleJson)},
|
||||
action_json = ${sql.json(actionJson)},
|
||||
concurrency_policy = ${concurrencyPolicy},
|
||||
next_run_at = ${nextRunAt},
|
||||
updated_at = now()
|
||||
WHERE id = ${scheduleId}
|
||||
RETURNING *
|
||||
`;
|
||||
await recordEvent("scheduled_task_updated", "scheduler", { scheduleId, name, enabled, schedule: scheduleJson, action: compactJson(action) });
|
||||
return jsonResponse({ ok: true, schedule: scheduledTaskView(rows[0]) });
|
||||
}
|
||||
|
||||
async function deleteScheduledTask(scheduleId: string): Promise<Response> {
|
||||
const rows = await sql<Array<{ id: string }>>`
|
||||
DELETE FROM unidesk_scheduled_tasks
|
||||
WHERE id = ${scheduleId}
|
||||
RETURNING id
|
||||
`;
|
||||
if (rows.length === 0) return jsonResponse({ ok: false, error: `scheduled task not found: ${scheduleId}` }, 404);
|
||||
await recordEvent("scheduled_task_deleted", "scheduler", { scheduleId });
|
||||
return jsonResponse({ ok: true, deleted: scheduleId });
|
||||
}
|
||||
|
||||
function scheduledRawTaskJson(task: RawTaskRow | null): JsonValue {
|
||||
if (task === null) return null;
|
||||
return {
|
||||
id: task.id,
|
||||
providerId: task.provider_id,
|
||||
command: task.command,
|
||||
status: task.status,
|
||||
payload: compactJson(task.payload),
|
||||
result: compactJson(task.result ?? null),
|
||||
updatedAt: rowIso(task.updated_at),
|
||||
};
|
||||
}
|
||||
|
||||
async function executeDispatchScheduleAction(action: DispatchScheduleAction): Promise<{ ok: boolean; taskId: string; result: JsonValue }> {
|
||||
const { taskId, providerOnline } = await createAndSendTask(action.providerId, action.command, {
|
||||
source: "scheduled-task",
|
||||
...action.payload,
|
||||
});
|
||||
if (!providerOnline) {
|
||||
return { ok: false, taskId, result: { providerOnline, taskId, error: `provider is offline: ${action.providerId}` } };
|
||||
}
|
||||
const timeoutMs = action.timeoutMs ?? numberInRange(action.payload.timeoutMs, config.taskPendingTimeoutMs + 5000, 1000, 24 * 3600_000);
|
||||
const task = await waitForTaskTerminal(taskId, timeoutMs);
|
||||
const ok = task?.status === "succeeded";
|
||||
return {
|
||||
ok,
|
||||
taskId,
|
||||
result: {
|
||||
providerOnline,
|
||||
taskId,
|
||||
timeoutMs,
|
||||
terminal: task === null ? false : isTerminalTaskStatus(task.status),
|
||||
task: scheduledRawTaskJson(task),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeRemoteDir(input: string): string {
|
||||
const raw = input.trim() || "/";
|
||||
if (raw.includes("\0")) throw new Error("remote directory must not contain null bytes");
|
||||
const normalized = pathPosix.normalize(raw.startsWith("/") ? raw : `/${raw}`);
|
||||
return normalized === "/" ? "/" : normalized.replace(/\/+$/u, "");
|
||||
}
|
||||
|
||||
function normalizeRelativeStagingPath(input: string): { relativePath: string; absolutePath: string } {
|
||||
const cleaned = input.replace(/\\/g, "/").replace(/^\/+/u, "").trim();
|
||||
const normalized = pathPosix.normalize(cleaned || ".");
|
||||
if (normalized === "." || normalized.startsWith("../") || normalized === "..") throw new Error("staging path must be a relative child path");
|
||||
const absolutePath = resolve(config.pgdataBackupStagingDir, normalized);
|
||||
const rel = relative(config.pgdataBackupStagingDir, absolutePath);
|
||||
if (rel.startsWith("..") || rel.includes("\0") || resolve(absolutePath) === resolve(config.pgdataBackupStagingDir)) {
|
||||
throw new Error("staging path must stay inside PGDATA_BACKUP_STAGING_DIR");
|
||||
}
|
||||
return { relativePath: normalized, absolutePath };
|
||||
}
|
||||
|
||||
function remoteFolderChain(remoteDir: string): string[] {
|
||||
const parts = normalizeRemoteDir(remoteDir).split("/").filter(Boolean);
|
||||
const folders: string[] = [];
|
||||
let current = "";
|
||||
for (const part of parts) {
|
||||
current = `${current}/${part}`;
|
||||
folders.push(current);
|
||||
}
|
||||
return folders;
|
||||
}
|
||||
|
||||
function safeFilePart(value: string): string {
|
||||
return value.replace(/[^A-Za-z0-9_.-]+/g, "_").replace(/^_+|_+$/g, "").slice(0, 80) || "data";
|
||||
}
|
||||
|
||||
function utcBackupParts(date = new Date()): { date: string; time: string; month: string; stamp: string } {
|
||||
const year = String(date.getUTCFullYear()).padStart(4, "0");
|
||||
const monthNumber = String(date.getUTCMonth() + 1).padStart(2, "0");
|
||||
const day = String(date.getUTCDate()).padStart(2, "0");
|
||||
const hour = String(date.getUTCHours()).padStart(2, "0");
|
||||
const minute = String(date.getUTCMinutes()).padStart(2, "0");
|
||||
const second = String(date.getUTCSeconds()).padStart(2, "0");
|
||||
return { date: `${year}${monthNumber}${day}`, time: `${hour}${minute}${second}`, month: `${year}${monthNumber}`, stamp: `${year}${monthNumber}${day}_${hour}${minute}${second}` };
|
||||
}
|
||||
|
||||
function databaseConnectionParts(): { host: string; port: string; user: string; password: string } {
|
||||
const url = new URL(config.databaseUrl);
|
||||
return {
|
||||
host: url.hostname || "database",
|
||||
port: url.port || "5432",
|
||||
user: decodeURIComponent(url.username || "postgres"),
|
||||
password: decodeURIComponent(url.password || ""),
|
||||
};
|
||||
}
|
||||
|
||||
async function runLocalCommand(
|
||||
command: string,
|
||||
args: string[],
|
||||
options: { cwd?: string; env?: Record<string, string>; timeoutMs: number },
|
||||
): Promise<{ ok: boolean; stdout: string; stderr: string; exitCode: number | null; timedOut: boolean }> {
|
||||
const proc = Bun.spawn([command, ...args], {
|
||||
cwd: options.cwd,
|
||||
env: { ...process.env, ...(options.env ?? {}) },
|
||||
stdout: "pipe",
|
||||
stderr: "pipe",
|
||||
});
|
||||
let timedOut = false;
|
||||
const timer = setTimeout(() => {
|
||||
timedOut = true;
|
||||
proc.kill("SIGTERM");
|
||||
}, Math.max(1, options.timeoutMs));
|
||||
try {
|
||||
const [stdout, stderr, exitCode] = await Promise.all([
|
||||
new Response(proc.stdout).text(),
|
||||
new Response(proc.stderr).text(),
|
||||
proc.exited,
|
||||
]);
|
||||
return { ok: exitCode === 0 && !timedOut, stdout: truncateText(stdout, 4000), stderr: truncateText(stderr, 4000), exitCode, timedOut };
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
async function sha256File(filePath: string): Promise<string> {
|
||||
const hash = createHash("sha256");
|
||||
await new Promise<void>((resolveHash, rejectHash) => {
|
||||
const stream = createReadStream(filePath);
|
||||
stream.on("data", (chunk) => hash.update(chunk));
|
||||
stream.on("error", rejectHash);
|
||||
stream.on("end", resolveHash);
|
||||
});
|
||||
return hash.digest("hex");
|
||||
}
|
||||
|
||||
async function fetchJsonWithTimeout(url: string, init: RequestInit, timeoutMs: number): Promise<Record<string, unknown>> {
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), Math.max(1, timeoutMs));
|
||||
try {
|
||||
const response = await fetch(url, { ...init, signal: controller.signal });
|
||||
const text = await response.text();
|
||||
let parsed: unknown = {};
|
||||
try {
|
||||
parsed = text.length > 0 ? JSON.parse(text) as unknown : {};
|
||||
} catch {
|
||||
parsed = { text };
|
||||
}
|
||||
if (!response.ok) throw new Error(`HTTP ${response.status}: ${truncateText(text, 1000)}`);
|
||||
if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) return { value: parsed };
|
||||
return parsed as Record<string, unknown>;
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
async function baiduJson(baseUrl: string, path: string, init: RequestInit = {}, timeoutMs = 30_000): Promise<Record<string, unknown>> {
|
||||
const url = new URL(path, baseUrl);
|
||||
const headers = new Headers(init.headers);
|
||||
if (init.body !== undefined && !headers.has("content-type")) headers.set("content-type", "application/json");
|
||||
const body = await fetchJsonWithTimeout(url.toString(), { ...init, headers }, timeoutMs);
|
||||
if (body.ok === false) throw new Error(`Baidu Netdisk API failed: ${JSON.stringify(compactJson(body))}`);
|
||||
return body;
|
||||
}
|
||||
|
||||
function jobRecord(value: unknown): Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
}
|
||||
|
||||
async function waitForBaiduTransfer(baseUrl: string, jobId: string, timeoutMs: number): Promise<Record<string, unknown>> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
let latest: Record<string, unknown> = {};
|
||||
while (Date.now() < deadline) {
|
||||
const detail = await baiduJson(baseUrl, `/api/transfers/${encodeURIComponent(jobId)}`, {}, 30_000);
|
||||
latest = detail;
|
||||
const job = jobRecord(detail.job);
|
||||
const status = String(job.status || "");
|
||||
if (status === "succeeded") return detail;
|
||||
if (status === "failed" || status === "canceled") throw new Error(`Baidu transfer ${status}: ${String(job.error || "no error detail")}`);
|
||||
await Bun.sleep(2000);
|
||||
}
|
||||
throw new Error(`Baidu transfer timed out after ${timeoutMs}ms: ${JSON.stringify(compactJson(latest))}`);
|
||||
}
|
||||
|
||||
async function executePgdataBackupAction(action: PgdataBackupScheduleAction, runId: string): Promise<{ ok: boolean; result: JsonValue }> {
|
||||
const startedAt = new Date();
|
||||
const parts = utcBackupParts(startedAt);
|
||||
const filename = `${parts.stamp}_${safeFilePart(action.volumeName)}.pg_basebackup.tar.gz`;
|
||||
const monthRelativeDir = pathPosix.join(action.stagingSubdir, parts.month);
|
||||
const backupRelativePath = pathPosix.join(monthRelativeDir, filename);
|
||||
const backupPath = normalizeRelativeStagingPath(backupRelativePath).absolutePath;
|
||||
const tempRelativeDir = pathPosix.join(action.stagingSubdir, parts.month, `.tmp_${runId}`);
|
||||
const tempDir = normalizeRelativeStagingPath(tempRelativeDir).absolutePath;
|
||||
const remoteDir = normalizeRemoteDir(pathPosix.join(action.remoteBaseDir, parts.month));
|
||||
const remotePath = normalizeRemoteDir(pathPosix.join(remoteDir, filename));
|
||||
const db = databaseConnectionParts();
|
||||
await mkdir(dirname(backupPath), { recursive: true });
|
||||
await rm(tempDir, { recursive: true, force: true });
|
||||
await mkdir(tempDir, { recursive: true });
|
||||
let uploadDetail: Record<string, unknown> | null = null;
|
||||
let backupBytes = 0;
|
||||
let backupSha256 = "";
|
||||
try {
|
||||
const backupTimeoutMs = Math.max(60_000, Math.floor(action.timeoutMs * 0.55));
|
||||
const packageTimeoutMs = Math.max(60_000, Math.floor(action.timeoutMs * 0.15));
|
||||
const uploadTimeoutMs = Math.max(60_000, action.timeoutMs - backupTimeoutMs - packageTimeoutMs);
|
||||
const basebackup = await runLocalCommand("pg_basebackup", [
|
||||
"-h", db.host,
|
||||
"-p", db.port,
|
||||
"-U", db.user,
|
||||
"-D", tempDir,
|
||||
"-Ft",
|
||||
"-X", "stream",
|
||||
"-z",
|
||||
"--checkpoint=fast",
|
||||
"--no-sync",
|
||||
], { env: { PGPASSWORD: db.password }, timeoutMs: backupTimeoutMs });
|
||||
if (!basebackup.ok) throw new Error(`pg_basebackup failed exit=${basebackup.exitCode} timedOut=${basebackup.timedOut}: ${basebackup.stderr || basebackup.stdout}`);
|
||||
const packaged = await runLocalCommand("tar", ["-czf", backupPath, "-C", tempDir, "."], { timeoutMs: packageTimeoutMs });
|
||||
if (!packaged.ok) throw new Error(`tar packaging failed exit=${packaged.exitCode} timedOut=${packaged.timedOut}: ${packaged.stderr || packaged.stdout}`);
|
||||
const info = await stat(backupPath);
|
||||
backupBytes = info.size;
|
||||
backupSha256 = await sha256File(backupPath);
|
||||
for (const folder of remoteFolderChain(remoteDir)) {
|
||||
await baiduJson(action.baiduBaseUrl, "/api/folders", {
|
||||
method: "POST",
|
||||
body: JSON.stringify({ path: folder }),
|
||||
}, 60_000);
|
||||
}
|
||||
const created = await baiduJson(action.baiduBaseUrl, "/api/transfers/upload-from-path", {
|
||||
method: "POST",
|
||||
body: JSON.stringify({ localPath: backupRelativePath, remotePath }),
|
||||
}, 60_000);
|
||||
const jobId = String(jobRecord(created.job).id || "");
|
||||
if (jobId.length === 0) throw new Error(`Baidu upload did not return a job id: ${JSON.stringify(compactJson(created))}`);
|
||||
uploadDetail = await waitForBaiduTransfer(action.baiduBaseUrl, jobId, uploadTimeoutMs);
|
||||
const uploadedJob = jobRecord(uploadDetail.job);
|
||||
return {
|
||||
ok: true,
|
||||
result: {
|
||||
backupType: "pg_basebackup",
|
||||
volumeName: action.volumeName,
|
||||
backupBytes,
|
||||
backupSha256,
|
||||
localRelativePath: backupRelativePath,
|
||||
remotePath,
|
||||
remoteDir,
|
||||
month: parts.month,
|
||||
timestampPrefix: parts.stamp,
|
||||
baiduTransferJobId: jobId,
|
||||
baiduTransferStatus: String(uploadedJob.status || ""),
|
||||
baiduFsId: String(uploadedJob.fsId || ""),
|
||||
baiduResult: compactJson(uploadedJob.result ?? null),
|
||||
},
|
||||
};
|
||||
} finally {
|
||||
await rm(tempDir, { recursive: true, force: true }).catch(() => undefined);
|
||||
if (action.cleanupLocal) await rm(backupPath, { force: true }).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
async function executeScheduleAction(action: ScheduleAction, runId: string): Promise<{ ok: boolean; taskId: string | null; result: JsonValue }> {
|
||||
if (action.type === "dispatch") {
|
||||
const dispatched = await executeDispatchScheduleAction(action);
|
||||
return { ok: dispatched.ok, taskId: dispatched.taskId, result: dispatched.result };
|
||||
}
|
||||
const backup = await executePgdataBackupAction(action, runId);
|
||||
return { ok: backup.ok, taskId: null, result: backup.result };
|
||||
}
|
||||
|
||||
async function executeScheduledRun(runId: string): Promise<void> {
|
||||
if (activeScheduledRuns.has(runId)) return;
|
||||
activeScheduledRuns.add(runId);
|
||||
const started = Date.now();
|
||||
try {
|
||||
const queuedRuns = await sql<ScheduledTaskRunRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_task_runs
|
||||
WHERE id = ${runId}
|
||||
LIMIT 1
|
||||
`;
|
||||
const queuedRun = queuedRuns[0];
|
||||
if (queuedRun === undefined) return;
|
||||
const scheduleRow = await getScheduledTaskRow(queuedRun.schedule_id);
|
||||
if (scheduleRow === null) return;
|
||||
const runRows = await sql<ScheduledTaskRunRow[]>`
|
||||
UPDATE unidesk_scheduled_task_runs
|
||||
SET status = 'running', started_at = now(), updated_at = now()
|
||||
WHERE id = ${runId} AND status = 'queued'
|
||||
RETURNING *
|
||||
`;
|
||||
if (runRows.length === 0) return;
|
||||
try {
|
||||
const action = normalizeScheduleAction(scheduleRow.action_json);
|
||||
const outcome = await executeScheduleAction(action, runId);
|
||||
const durationMs = Date.now() - started;
|
||||
const status = outcome.ok ? "succeeded" : "failed";
|
||||
const error = outcome.ok ? null : "scheduled action failed";
|
||||
await sql.begin(async (tx) => {
|
||||
await tx`
|
||||
UPDATE unidesk_scheduled_task_runs
|
||||
SET status = ${status},
|
||||
task_id = ${outcome.taskId},
|
||||
result = ${tx.json(outcome.result)},
|
||||
error = ${error},
|
||||
finished_at = now(),
|
||||
duration_ms = ${durationMs},
|
||||
updated_at = now()
|
||||
WHERE id = ${runId}
|
||||
`;
|
||||
await tx`
|
||||
UPDATE unidesk_scheduled_tasks
|
||||
SET last_run_at = now(), last_run_id = ${runId}, updated_at = now()
|
||||
WHERE id = ${scheduleRow.id}
|
||||
`;
|
||||
});
|
||||
await recordEvent(`scheduled_task_${status}`, "scheduler", { scheduleId: scheduleRow.id, runId, durationMs, result: compactJson(outcome.result) });
|
||||
} catch (error) {
|
||||
const durationMs = Date.now() - started;
|
||||
await sql.begin(async (tx) => {
|
||||
await tx`
|
||||
UPDATE unidesk_scheduled_task_runs
|
||||
SET status = 'failed',
|
||||
result = ${tx.json(errorToJson(error))},
|
||||
error = ${error instanceof Error ? error.message : String(error)},
|
||||
finished_at = now(),
|
||||
duration_ms = ${durationMs},
|
||||
updated_at = now()
|
||||
WHERE id = ${runId}
|
||||
`;
|
||||
await tx`
|
||||
UPDATE unidesk_scheduled_tasks
|
||||
SET last_run_at = now(), last_run_id = ${runId}, updated_at = now()
|
||||
WHERE id = ${scheduleRow.id}
|
||||
`;
|
||||
});
|
||||
await recordEvent("scheduled_task_failed", "scheduler", { scheduleId: scheduleRow.id, runId, durationMs, error: errorToJson(error) });
|
||||
}
|
||||
} finally {
|
||||
activeScheduledRuns.delete(runId);
|
||||
}
|
||||
}
|
||||
|
||||
async function triggerScheduledTask(scheduleId: string, triggerType: "manual" | "schedule"): Promise<JsonValue | null> {
|
||||
const runId = `schedrun_${Date.now()}_${Math.random().toString(16).slice(2, 8)}`;
|
||||
let createdRun: ScheduledTaskRunRow | null = null;
|
||||
await sql.begin(async (tx) => {
|
||||
const rows = await tx<ScheduledTaskRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_tasks
|
||||
WHERE id = ${scheduleId}
|
||||
FOR UPDATE
|
||||
`;
|
||||
const scheduleRow = rows[0];
|
||||
if (scheduleRow === undefined) throw new Error(`scheduled task not found: ${scheduleId}`);
|
||||
if (triggerType === "schedule" && !scheduleRow.enabled) return;
|
||||
const runningRows = await tx<Array<{ count: string | number }>>`
|
||||
SELECT count(*)::int AS count
|
||||
FROM unidesk_scheduled_task_runs
|
||||
WHERE schedule_id = ${scheduleId}
|
||||
AND status IN ('queued', 'running')
|
||||
`;
|
||||
const runningCount = Number(runningRows[0]?.count ?? 0);
|
||||
const schedule = normalizeScheduleSpec(scheduleRow.schedule_json);
|
||||
const shouldAdvanceNext = triggerType === "schedule";
|
||||
if (shouldAdvanceNext) {
|
||||
await tx`
|
||||
UPDATE unidesk_scheduled_tasks
|
||||
SET next_run_at = ${computeNextRunAt(schedule).toISOString()}, updated_at = now()
|
||||
WHERE id = ${scheduleId}
|
||||
`;
|
||||
}
|
||||
if (scheduleRow.concurrency_policy !== "parallel" && runningCount > 0) {
|
||||
const skippedRows = await tx<ScheduledTaskRunRow[]>`
|
||||
INSERT INTO unidesk_scheduled_task_runs (id, schedule_id, trigger_type, status, result, error, finished_at, duration_ms)
|
||||
VALUES (${runId}, ${scheduleId}, ${triggerType}, 'skipped', ${tx.json({ reason: "previous run still active", runningCount })}, 'previous run still active', now(), 0)
|
||||
RETURNING *
|
||||
`;
|
||||
createdRun = skippedRows[0] ?? null;
|
||||
return;
|
||||
}
|
||||
const runRows = await tx<ScheduledTaskRunRow[]>`
|
||||
INSERT INTO unidesk_scheduled_task_runs (id, schedule_id, trigger_type, status)
|
||||
VALUES (${runId}, ${scheduleId}, ${triggerType}, 'queued')
|
||||
RETURNING *
|
||||
`;
|
||||
createdRun = runRows[0] ?? null;
|
||||
});
|
||||
const run = createdRun as ScheduledTaskRunRow | null;
|
||||
if (run === null) return null;
|
||||
if (run.status === "queued") {
|
||||
setTimeout(() => {
|
||||
executeScheduledRun(runId).catch((error) => logger("error", "scheduled_run_uncaught", { runId, error: errorToJson(error) }));
|
||||
}, 0);
|
||||
}
|
||||
await recordEvent("scheduled_task_triggered", "scheduler", { scheduleId, runId, triggerType, status: run.status });
|
||||
return scheduleRunView(run);
|
||||
}
|
||||
|
||||
async function scheduledTaskRoute(req: Request, url: URL): Promise<Response> {
|
||||
const prefix = "/api/schedules";
|
||||
const rest = url.pathname === prefix ? "" : url.pathname.slice(prefix.length + 1);
|
||||
const segments = rest.split("/").filter(Boolean).map((part) => decodeURIComponent(part));
|
||||
if (url.pathname === prefix && req.method === "GET") {
|
||||
return jsonResponse({ ok: true, schedules: await getScheduledTasks(readLimit(url, 100)) });
|
||||
}
|
||||
if (url.pathname === prefix && req.method === "POST") return upsertScheduledTask(req, null);
|
||||
if (segments.length === 1 && segments[0] === "runs" && req.method === "GET") {
|
||||
return jsonResponse({ ok: true, runs: await getScheduledTaskRuns(null, readLimit(url, 100)) });
|
||||
}
|
||||
if (segments.length === 1 && req.method === "GET") {
|
||||
const schedule = await getScheduledTaskRow(segments[0]);
|
||||
if (schedule === null) return jsonResponse({ ok: false, error: `scheduled task not found: ${segments[0]}` }, 404);
|
||||
const runs = await sql<ScheduledTaskRunRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_task_runs
|
||||
WHERE schedule_id = ${segments[0]}
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT 20
|
||||
`;
|
||||
return jsonResponse({ ok: true, schedule: scheduledTaskView(schedule, runs) });
|
||||
}
|
||||
if (segments.length === 1 && req.method === "PUT") return upsertScheduledTask(req, segments[0]);
|
||||
if (segments.length === 1 && req.method === "PATCH") return patchScheduledTask(segments[0], req);
|
||||
if (segments.length === 1 && req.method === "DELETE") return deleteScheduledTask(segments[0]);
|
||||
if (segments.length === 2 && segments[1] === "run" && req.method === "POST") {
|
||||
const run = await triggerScheduledTask(segments[0], "manual");
|
||||
if (run === null) return jsonResponse({ ok: false, error: `scheduled task was not triggered: ${segments[0]}` }, 409);
|
||||
return jsonResponse({ ok: true, run });
|
||||
}
|
||||
if (segments.length === 2 && segments[1] === "runs" && req.method === "GET") {
|
||||
return jsonResponse({ ok: true, runs: await getScheduledTaskRuns(segments[0], readLimit(url, 100)) });
|
||||
}
|
||||
return jsonResponse({ ok: false, error: "scheduled task route not found", path: url.pathname }, 404);
|
||||
}
|
||||
|
||||
async function recoverScheduledRuns(): Promise<void> {
|
||||
const rows = await sql<ScheduledTaskRunRow[]>`
|
||||
UPDATE unidesk_scheduled_task_runs
|
||||
SET status = 'failed',
|
||||
error = 'backend-core restarted before scheduled run completed',
|
||||
result = jsonb_build_object('error', 'backend-core restarted before scheduled run completed'),
|
||||
finished_at = now(),
|
||||
updated_at = now()
|
||||
WHERE status IN ('queued', 'running')
|
||||
RETURNING *
|
||||
`;
|
||||
if (rows.length > 0) await recordEvent("scheduled_runs_recovered", "scheduler", { failedRunCount: rows.length });
|
||||
const schedules = await sql<ScheduledTaskRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_tasks
|
||||
WHERE next_run_at IS NULL
|
||||
LIMIT 200
|
||||
`;
|
||||
for (const schedule of schedules) {
|
||||
const nextRunAt = computeNextRunAt(normalizeScheduleSpec(schedule.schedule_json)).toISOString();
|
||||
await sql`UPDATE unidesk_scheduled_tasks SET next_run_at = ${nextRunAt}, updated_at = now() WHERE id = ${schedule.id}`;
|
||||
}
|
||||
}
|
||||
|
||||
async function runDueScheduledTasks(): Promise<void> {
|
||||
if (!dbReady) return;
|
||||
const rows = await sql<ScheduledTaskRow[]>`
|
||||
SELECT *
|
||||
FROM unidesk_scheduled_tasks
|
||||
WHERE enabled = true
|
||||
AND (next_run_at IS NULL OR next_run_at <= now())
|
||||
ORDER BY next_run_at ASC NULLS FIRST
|
||||
LIMIT 5
|
||||
`;
|
||||
for (const row of rows) {
|
||||
try {
|
||||
await triggerScheduledTask(row.id, "schedule");
|
||||
} catch (error) {
|
||||
logger("error", "scheduled_task_due_trigger_failed", { scheduleId: row.id, error: errorToJson(error) });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function isMicroservicePathAllowed(service: MicroserviceConfig, path: string): boolean {
|
||||
return service.backend.allowedPathPrefixes.some((prefix) => path === prefix || path.startsWith(prefix));
|
||||
}
|
||||
@@ -2228,6 +3044,9 @@ async function routeInner(req: Request, server: Server<WsData>): Promise<Respons
|
||||
const summary = ["1", "true", "yes"].includes((url.searchParams.get("summary") ?? "").toLowerCase());
|
||||
return jsonResponse({ ok: true, tasks: await withPerformanceOperation("core", "tasks", url.search, () => getTasks(readLimit(url, 100), url.searchParams.get("status") ?? "all", lite, summary)) });
|
||||
}
|
||||
if (url.pathname === "/api/schedules" || url.pathname.startsWith("/api/schedules/")) {
|
||||
return withPerformanceOperation("scheduler", "schedules", url.pathname, () => scheduledTaskRoute(req, url));
|
||||
}
|
||||
if (url.pathname === "/api/microservices") return jsonResponse({ ok: true, microservices: await withPerformanceOperation("core", "microservices", url.pathname, () => getMicroservices()) });
|
||||
if (url.pathname === "/api/performance") return jsonResponse(await getPerformance());
|
||||
if (url.pathname === "/api/code-queue-load-test" && (req.method === "GET" || req.method === "POST")) return withPerformanceOperation("performance", "code_queue_load_test", url.pathname, () => codexQueueLoadTest(req));
|
||||
@@ -2281,6 +3100,8 @@ function readLimit(url: URL, defaultLimit: number): number {
|
||||
|
||||
await initDatabaseWithRetry();
|
||||
markStaleTasksFailed().catch((error) => logger("error", "task_timeout_sweep_failed", { error: errorToJson(error) }));
|
||||
await recoverScheduledRuns();
|
||||
runDueScheduledTasks().catch((error) => logger("error", "scheduled_task_sweep_failed", { error: errorToJson(error) }));
|
||||
|
||||
const apiServer = Bun.serve<WsData>({
|
||||
port: config.port,
|
||||
@@ -2365,6 +3186,10 @@ setInterval(() => {
|
||||
markStaleTasksFailed().catch((error) => logger("error", "task_timeout_sweep_failed", { error: errorToJson(error) }));
|
||||
}, Math.min(config.taskPendingTimeoutMs, 60_000));
|
||||
|
||||
setInterval(() => {
|
||||
runDueScheduledTasks().catch((error) => logger("error", "scheduled_task_sweep_failed", { error: errorToJson(error) }));
|
||||
}, 30_000);
|
||||
|
||||
logger("info", "server_listening", {
|
||||
apiUrl: `http://0.0.0.0:${apiServer.port}`,
|
||||
providerIngressUrl: `ws://0.0.0.0:${providerServer.port}/ws/provider`,
|
||||
|
||||
Reference in New Issue
Block a user