feat(FN-1266): add dashboard API integration tests for budget endpoints

This commit is contained in:
gsxdsm
2026-04-10 01:05:21 -07:00
parent bb5a78f422
commit 818562bd95
10 changed files with 1520 additions and 14 deletions

View File

@@ -51,7 +51,7 @@ describe("TaskStore task documents", () => {
expect(tableNames.has("task_documents")).toBe(true);
expect(tableNames.has("task_document_revisions")).toBe(true);
expect(db.getSchemaVersion()).toBe(26);
expect(db.getSchemaVersion()).toBe(27);
const index = db
.prepare(

View File

@@ -106,7 +106,7 @@ describe("Database", () => {
});
it("seeds schema version", () => {
expect(db.getSchemaVersion()).toBe(26);
expect(db.getSchemaVersion()).toBe(27);
});
it("seeds lastModified", () => {
@@ -129,7 +129,7 @@ describe("Database", () => {
it("is idempotent - calling init() twice does not fail", () => {
expect(() => db.init()).not.toThrow();
expect(db.getSchemaVersion()).toBe(26);
expect(db.getSchemaVersion()).toBe(27);
});
it("does not overwrite existing config on re-init", () => {
@@ -735,8 +735,8 @@ describe("schema migrations", () => {
// Now run init() which should trigger migration
db.init();
// Verify version bumped to 22 (includes v1→v2 through v21→v22)
expect(db.getSchemaVersion()).toBe(26);
// Verify version bumped to 27 (includes v1→v2 through v26→v27)
expect(db.getSchemaVersion()).toBe(27);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -761,11 +761,11 @@ describe("schema migrations", () => {
const db = new Database(kbDir);
db.init();
expect(db.getSchemaVersion()).toBe(26);
expect(db.getSchemaVersion()).toBe(27);
// Re-init should not fail
db.init();
expect(db.getSchemaVersion()).toBe(26);
expect(db.getSchemaVersion()).toBe(27);
db.close();
});
@@ -781,7 +781,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(26);
expect(db.getSchemaVersion()).toBe(27);
const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'agentRatings'").all() as Array<{ name: string }>;
expect(tables).toEqual([{ name: "agentRatings" }]);
@@ -805,7 +805,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(26);
expect(db.getSchemaVersion()).toBe(27);
const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'mission_events'").all() as Array<{ name: string }>;
expect(tables).toEqual([{ name: "mission_events" }]);
@@ -908,8 +908,8 @@ describe("schema migrations", () => {
// Now run init() which should trigger migrations v2→v3→v4
db.init();
// Verify version bumped to 22
expect(db.getSchemaVersion()).toBe(26);
// Verify version bumped to 27
expect(db.getSchemaVersion()).toBe(27);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -1275,7 +1275,7 @@ describe("createDatabase factory", () => {
const db = createDatabase(kbDir);
db.init();
expect(db.getSchemaVersion()).toBe(26);
expect(db.getSchemaVersion()).toBe(27);
expect(db.getLastModified()).toBeGreaterThan(0);
db.close();

View File

@@ -59,7 +59,7 @@ export function fromJson<T>(json: string | null | undefined): T | undefined {
// ── Schema Definition ────────────────────────────────────────────────
const SCHEMA_VERSION = 26;
const SCHEMA_VERSION = 27;
function normalizeTaskComments(
steeringComments: SteeringComment[] | undefined,
@@ -393,6 +393,27 @@ CREATE TABLE IF NOT EXISTS plugins (
createdAt TEXT NOT NULL,
updatedAt TEXT NOT NULL
);
-- Routines table for recurring task automation
CREATE TABLE IF NOT EXISTS routines (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
description TEXT,
triggerType TEXT NOT NULL,
triggerConfig TEXT NOT NULL,
catchUpPolicy TEXT NOT NULL DEFAULT 'run_one',
executionPolicy TEXT NOT NULL DEFAULT 'queue',
enabled INTEGER DEFAULT 1,
lastRunAt TEXT,
lastRunResult TEXT,
nextRunAt TEXT,
runCount INTEGER DEFAULT 0,
runHistory TEXT DEFAULT '[]',
createdAt TEXT NOT NULL,
updatedAt TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idxRoutinesNextRunAt ON routines(nextRunAt);
CREATE INDEX IF NOT EXISTS idxRoutinesEnabled ON routines(enabled);
`;
// ── Database Class ───────────────────────────────────────────────────
@@ -943,6 +964,32 @@ export class Database {
this.db.exec(`CREATE INDEX IF NOT EXISTS idxTasksAssigneeUserId ON tasks(assigneeUserId)`);
});
}
if (version < 27) {
this.applyMigration(27, () => {
this.db.exec(`
CREATE TABLE IF NOT EXISTS routines (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
description TEXT,
triggerType TEXT NOT NULL,
triggerConfig TEXT NOT NULL,
catchUpPolicy TEXT NOT NULL DEFAULT 'run_one',
executionPolicy TEXT NOT NULL DEFAULT 'queue',
enabled INTEGER DEFAULT 1,
lastRunAt TEXT,
lastRunResult TEXT,
nextRunAt TEXT,
runCount INTEGER DEFAULT 0,
runHistory TEXT DEFAULT '[]',
createdAt TEXT NOT NULL,
updatedAt TEXT NOT NULL
)
`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxRoutinesNextRunAt ON routines(nextRunAt)`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxRoutinesEnabled ON routines(enabled)`);
});
}
}
/**

View File

@@ -66,6 +66,31 @@ export type { ScheduleType, ScheduledTask, ScheduledTaskCreateInput, ScheduledTa
export { AutomationStore } from "./automation-store.js";
export type { AutomationStoreEvents } from "./automation-store.js";
// ── Routine System ───────────────────────────────────────────────────
export {
MAX_ROUTINE_RUN_HISTORY,
isCronTrigger,
isWebhookTrigger,
isApiTrigger,
isManualTrigger,
} from "./routine.js";
export type {
RoutineTriggerType,
RoutineCronTrigger,
RoutineWebhookTrigger,
RoutineApiTrigger,
RoutineManualTrigger,
RoutineTrigger,
RoutineCatchUpPolicy,
RoutineExecutionPolicy,
RoutineExecutionResult,
Routine,
RoutineCreateInput,
RoutineUpdateInput,
} from "./routine.js";
export { RoutineStore } from "./routine-store.js";
export type { RoutineStoreEvents } from "./routine-store.js";
// ── Plugin System ─────────────────────────────────────────────────────
export type {
PluginManifest,

View File

@@ -0,0 +1,567 @@
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { RoutineStore } from "./routine-store.js";
import { rm } from "node:fs/promises";
import { join } from "node:path";
import { mkdtempSync, existsSync } from "node:fs";
import { tmpdir } from "node:os";
import type {
Routine,
RoutineCreateInput,
RoutineExecutionResult,
RoutineTrigger,
} from "./routine.js";
function makeTmpDir(): string {
return mkdtempSync(join(tmpdir(), "kb-routine-test-"));
}
describe("RoutineStore", () => {
let rootDir: string;
let store: RoutineStore;
beforeEach(async () => {
rootDir = makeTmpDir();
store = new RoutineStore(rootDir);
await store.init();
});
afterEach(async () => {
await rm(rootDir, { recursive: true, force: true });
});
// ── init ──────────────────────────────────────────────────────────
describe("init", () => {
it("is idempotent", async () => {
await store.init();
await store.init();
// Should not throw
});
});
// ── isValidCron ─────────────────────────────────────────────────
describe("isValidCron", () => {
it("accepts valid cron expressions", () => {
expect(RoutineStore.isValidCron("0 * * * *")).toBe(true);
expect(RoutineStore.isValidCron("*/5 * * * *")).toBe(true);
expect(RoutineStore.isValidCron("0 0 * * 1")).toBe(true);
expect(RoutineStore.isValidCron("0 9 1 * *")).toBe(true);
});
it("rejects invalid cron expressions", () => {
expect(RoutineStore.isValidCron("not a cron")).toBe(false);
expect(RoutineStore.isValidCron("60 * * * *")).toBe(false);
expect(RoutineStore.isValidCron("0 25 * * *")).toBe(false);
});
});
// ── computeNextRun ────────────────────────────────────────────────
describe("computeNextRun", () => {
it("returns a future ISO timestamp", () => {
const fromDate = new Date("2026-01-01T00:00:00Z");
const next = store.computeNextRun("0 * * * *", fromDate);
expect(new Date(next).getTime()).toBeGreaterThan(fromDate.getTime());
});
it("computes correct next run for hourly", () => {
const fromDate = new Date("2026-01-01T12:30:00Z");
const next = store.computeNextRun("0 * * * *", fromDate);
expect(new Date(next).getUTCHours()).toBe(13);
expect(new Date(next).getUTCMinutes()).toBe(0);
});
});
// ── createRoutine ────────────────────────────────────────────────
describe("createRoutine", () => {
it("creates a routine with cron trigger", async () => {
const input: RoutineCreateInput = {
name: "Hourly check",
trigger: { type: "cron", cronExpression: "0 * * * *" },
};
const routine = await store.createRoutine(input);
expect(routine.id).toBeTruthy();
expect(routine.name).toBe("Hourly check");
expect(routine.trigger.type).toBe("cron");
expect((routine.trigger as any).cronExpression).toBe("0 * * * *");
expect(routine.catchUpPolicy).toBe("run_one");
expect(routine.executionPolicy).toBe("queue");
expect(routine.enabled).toBe(true);
expect(routine.runCount).toBe(0);
expect(routine.runHistory).toEqual([]);
expect(routine.nextRunAt).toBeTruthy();
expect(routine.createdAt).toBeTruthy();
expect(routine.updatedAt).toBeTruthy();
});
it("creates a routine with webhook trigger", async () => {
const input: RoutineCreateInput = {
name: "Webhook routine",
trigger: { type: "webhook", webhookPath: "/trigger/my-routine" },
};
const routine = await store.createRoutine(input);
expect(routine.trigger.type).toBe("webhook");
expect((routine.trigger as any).webhookPath).toBe("/trigger/my-routine");
});
it("creates a routine with api trigger", async () => {
const input: RoutineCreateInput = {
name: "API routine",
trigger: { type: "api", endpoint: "/api/routines/run" },
};
const routine = await store.createRoutine(input);
expect(routine.trigger.type).toBe("api");
expect((routine.trigger as any).endpoint).toBe("/api/routines/run");
});
it("creates a routine with manual trigger", async () => {
const input: RoutineCreateInput = {
name: "Manual routine",
trigger: { type: "manual" },
};
const routine = await store.createRoutine(input);
expect(routine.trigger.type).toBe("manual");
expect(routine.nextRunAt).toBeUndefined(); // No nextRunAt for manual triggers
});
it("creates disabled routine without nextRunAt", async () => {
const input: RoutineCreateInput = {
name: "Disabled",
trigger: { type: "cron", cronExpression: "0 * * * *" },
enabled: false,
};
const routine = await store.createRoutine(input);
expect(routine.enabled).toBe(false);
expect(routine.nextRunAt).toBeUndefined();
});
it("creates routine with custom policies", async () => {
const input: RoutineCreateInput = {
name: "Custom policies",
trigger: { type: "cron", cronExpression: "0 * * * *" },
catchUpPolicy: "skip",
executionPolicy: "parallel",
};
const routine = await store.createRoutine(input);
expect(routine.catchUpPolicy).toBe("skip");
expect(routine.executionPolicy).toBe("parallel");
});
it("rejects empty name", async () => {
const input: RoutineCreateInput = {
name: "",
trigger: { type: "manual" },
};
await expect(store.createRoutine(input)).rejects.toThrow("Name is required");
});
it("rejects invalid cron expression", async () => {
const input: RoutineCreateInput = {
name: "Bad cron",
trigger: { type: "cron", cronExpression: "bad cron" },
};
await expect(store.createRoutine(input)).rejects.toThrow("Invalid cron expression");
});
it("emits routine:created event", async () => {
const listener = vi.fn();
store.on("routine:created", listener);
const routine = await store.createRoutine({
name: "Event test",
trigger: { type: "manual" },
});
expect(listener).toHaveBeenCalledWith(routine);
});
});
// ── getRoutine ──────────────────────────────────────────────────
describe("getRoutine", () => {
it("reads a routine by id", async () => {
const created = await store.createRoutine({
name: "Get test",
trigger: { type: "manual" },
});
const fetched = await store.getRoutine(created.id);
expect(fetched.id).toBe(created.id);
expect(fetched.name).toBe("Get test");
});
it("throws ENOENT for missing routine", async () => {
await expect(store.getRoutine("nonexistent")).rejects.toThrow("not found");
});
});
// ── listRoutines ─────────────────────────────────────────────────
describe("listRoutines", () => {
it("returns empty array when no routines", async () => {
const list = await store.listRoutines();
expect(list).toEqual([]);
});
it("returns all routines sorted by createdAt", async () => {
await store.createRoutine({ name: "A", trigger: { type: "manual" } });
await new Promise((r) => setTimeout(r, 5));
await store.createRoutine({ name: "B", trigger: { type: "manual" } });
const list = await store.listRoutines();
expect(list).toHaveLength(2);
expect(list[0].name).toBe("A");
expect(list[1].name).toBe("B");
});
});
// ── updateRoutine ────────────────────────────────────────────────
describe("updateRoutine", () => {
it("updates name and description", async () => {
const routine = await store.createRoutine({
name: "Original",
trigger: { type: "manual" },
});
await new Promise((r) => setTimeout(r, 5));
const updated = await store.updateRoutine(routine.id, {
name: "Updated",
description: "A description",
});
expect(updated.name).toBe("Updated");
expect(updated.description).toBe("A description");
expect(new Date(updated.updatedAt).getTime()).toBeGreaterThanOrEqual(
new Date(routine.updatedAt).getTime(),
);
});
it("updates trigger from manual to cron", async () => {
const routine = await store.createRoutine({
name: "Test",
trigger: { type: "manual" },
});
const updated = await store.updateRoutine(routine.id, {
trigger: { type: "cron", cronExpression: "*/10 * * * *" },
});
expect(updated.trigger.type).toBe("cron");
expect((updated.trigger as any).cronExpression).toBe("*/10 * * * *");
expect(updated.nextRunAt).toBeTruthy();
});
it("updates enabled state", async () => {
const routine = await store.createRoutine({
name: "Toggle",
trigger: { type: "cron", cronExpression: "0 * * * *" },
});
const disabled = await store.updateRoutine(routine.id, { enabled: false });
expect(disabled.enabled).toBe(false);
expect(disabled.nextRunAt).toBeUndefined();
const reenabled = await store.updateRoutine(routine.id, { enabled: true });
expect(reenabled.enabled).toBe(true);
expect(reenabled.nextRunAt).toBeTruthy();
});
it("updates policies", async () => {
const routine = await store.createRoutine({
name: "Policies",
trigger: { type: "manual" },
});
const updated = await store.updateRoutine(routine.id, {
catchUpPolicy: "run",
executionPolicy: "parallel",
});
expect(updated.catchUpPolicy).toBe("run");
expect(updated.executionPolicy).toBe("parallel");
});
it("rejects empty name", async () => {
const routine = await store.createRoutine({
name: "Test",
trigger: { type: "manual" },
});
await expect(
store.updateRoutine(routine.id, { name: " " }),
).rejects.toThrow("Name cannot be empty");
});
it("rejects invalid cron on update", async () => {
const routine = await store.createRoutine({
name: "Test",
trigger: { type: "manual" },
});
await expect(
store.updateRoutine(routine.id, {
trigger: { type: "cron", cronExpression: "bad cron" },
}),
).rejects.toThrow("Invalid cron expression");
});
it("emits routine:updated event", async () => {
const routine = await store.createRoutine({
name: "Event test",
trigger: { type: "manual" },
});
const listener = vi.fn();
store.on("routine:updated", listener);
await store.updateRoutine(routine.id, { name: "Updated" });
expect(listener).toHaveBeenCalledTimes(1);
});
});
// ── deleteRoutine ───────────────────────────────────────────────
describe("deleteRoutine", () => {
it("deletes a routine", async () => {
const routine = await store.createRoutine({
name: "Delete me",
trigger: { type: "manual" },
});
const deleted = await store.deleteRoutine(routine.id);
expect(deleted.id).toBe(routine.id);
await expect(store.getRoutine(routine.id)).rejects.toThrow("not found");
});
it("throws for missing routine", async () => {
await expect(store.deleteRoutine("nonexistent")).rejects.toThrow("not found");
});
it("emits routine:deleted event", async () => {
const routine = await store.createRoutine({
name: "Delete test",
trigger: { type: "manual" },
});
const listener = vi.fn();
store.on("routine:deleted", listener);
await store.deleteRoutine(routine.id);
expect(listener).toHaveBeenCalledWith(routine);
});
});
// ── recordRun ───────────────────────────────────────────────────
describe("recordRun", () => {
it("records a successful run", async () => {
const routine = await store.createRoutine({
name: "Run test",
trigger: { type: "manual" },
});
const result: RoutineExecutionResult = {
routineId: routine.id,
success: true,
output: "completed",
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
};
const updated = await store.recordRun(routine.id, result);
expect(updated.lastRunAt).toBe(result.startedAt);
expect(updated.lastRunResult).toEqual(result);
expect(updated.runCount).toBe(1);
expect(updated.runHistory).toHaveLength(1);
expect(updated.runHistory[0]).toEqual(result);
});
it("records a failed run", async () => {
const routine = await store.createRoutine({
name: "Fail test",
trigger: { type: "manual" },
});
const result: RoutineExecutionResult = {
routineId: routine.id,
success: false,
output: "",
error: "Something went wrong",
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
};
const updated = await store.recordRun(routine.id, result);
expect(updated.lastRunResult?.success).toBe(false);
expect(updated.lastRunResult?.error).toContain("Something went wrong");
expect(updated.runCount).toBe(1);
});
it("caps run history at MAX_ROUTINE_RUN_HISTORY", async () => {
const routine = await store.createRoutine({
name: "History test",
trigger: { type: "manual" },
});
for (let i = 0; i < 55; i++) {
await store.recordRun(routine.id, {
routineId: routine.id,
success: true,
output: `run ${i}`,
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
});
}
const updated = await store.getRoutine(routine.id);
expect(updated.runHistory.length).toBeLessThanOrEqual(50);
expect(updated.runCount).toBe(55);
});
it("emits routine:run event", async () => {
const routine = await store.createRoutine({
name: "Event test",
trigger: { type: "manual" },
});
const listener = vi.fn();
store.on("routine:run", listener);
const result: RoutineExecutionResult = {
routineId: routine.id,
success: true,
output: "ok",
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
};
await store.recordRun(routine.id, result);
expect(listener).toHaveBeenCalledTimes(1);
expect(listener.mock.calls[0][0].result).toEqual(result);
});
it("recomputes nextRunAt for cron routines after run", async () => {
// Use a cron that fires every minute to ensure different nextRunAt
const routine = await store.createRoutine({
name: "Cron run test",
trigger: { type: "cron", cronExpression: "0 * * * * *" },
});
const originalNextRun = routine.nextRunAt;
expect(originalNextRun).toBeTruthy();
// Wait a bit to ensure time passes
await new Promise((r) => setTimeout(r, 1000));
const result: RoutineExecutionResult = {
routineId: routine.id,
success: true,
output: "ok",
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
};
const updated = await store.recordRun(routine.id, result);
expect(updated.nextRunAt).toBeTruthy();
// nextRunAt should be updated (may be same or later depending on timing)
expect(updated.nextRunAt).not.toBeUndefined();
});
});
// ── getDueRoutines ──────────────────────────────────────────────
describe("getDueRoutines", () => {
it("returns empty array when no routines", async () => {
const due = await store.getDueRoutines();
expect(due).toEqual([]);
});
it("excludes disabled routines", async () => {
const routine = await store.createRoutine({
name: "Disabled test",
trigger: { type: "cron", cronExpression: "0 * * * *" },
enabled: false,
});
const due = await store.getDueRoutines();
expect(due.some((d) => d.id === routine.id)).toBe(false);
});
it("excludes routines with future nextRunAt", async () => {
const routine = await store.createRoutine({
name: "Future test",
trigger: { type: "cron", cronExpression: "0 * * * *" },
});
// nextRunAt is in the future by default
const due = await store.getDueRoutines();
expect(due.some((d) => d.id === routine.id)).toBe(false);
});
it("returns routines with past nextRunAt after manual update", async () => {
// Create routine with cron trigger
const routine = await store.createRoutine({
name: "Due test",
trigger: { type: "cron", cronExpression: "0 * * * *" },
});
// Manually set nextRunAt to the past by directly manipulating the database
// This tests the due-routine query logic
const pastDate = new Date(Date.now() - 60000).toISOString();
store["db"].prepare(
"UPDATE routines SET nextRunAt = ? WHERE id = ?"
).run(pastDate, routine.id);
// Now getDueRoutines should include it
const due = await store.getDueRoutines();
expect(due.some((d) => d.id === routine.id)).toBe(true);
});
});
// ── Concurrent write safety ─────────────────────────────────────
describe("concurrency", () => {
it("handles concurrent updates safely", async () => {
const routine = await store.createRoutine({
name: "Concurrent",
trigger: { type: "manual" },
});
// Fire multiple concurrent recordRun calls
const updates = Array.from({ length: 10 }, (_, i) =>
store.recordRun(routine.id, {
routineId: routine.id,
success: true,
output: `run ${i}`,
startedAt: new Date().toISOString(),
completedAt: new Date().toISOString(),
}),
);
await Promise.all(updates);
const final = await store.getRoutine(routine.id);
expect(final.runCount).toBe(10);
expect(final.runHistory).toHaveLength(10);
});
});
});

View File

@@ -0,0 +1,390 @@
/**
* RoutineStore: SQLite-backed store for Routine CRUD, run tracking, and due queries.
*
* Follows the AutomationStore pattern with:
* - Lazy DB initialization
* - Per-routine mutation locking via promise chains
* - Typed EventEmitter lifecycle events
*/
import { EventEmitter } from "node:events";
import { randomUUID } from "node:crypto";
import { CronExpressionParser } from "cron-parser";
import { Database, toJson, fromJson } from "./db.js";
import {
isCronTrigger,
type Routine,
type RoutineTrigger,
type RoutineCreateInput,
type RoutineUpdateInput,
type RoutineExecutionResult,
type RoutineTriggerType,
type RoutineCronTrigger,
type RoutineWebhookTrigger,
type RoutineApiTrigger,
type RoutineManualTrigger,
MAX_ROUTINE_RUN_HISTORY,
} from "./routine.js";
export interface RoutineStoreEvents {
"routine:created": [routine: Routine];
"routine:updated": [routine: Routine];
"routine:deleted": [routine: Routine];
"routine:run": [data: { routine: Routine; result: RoutineExecutionResult }];
}
export class RoutineStore extends EventEmitter<RoutineStoreEvents> {
/** SQLite database instance (lazy init). */
private _db: Database | null = null;
/** Per-routine promise chain for serializing writes. */
private routineLocks: Map<string, Promise<void>> = new Map();
constructor(private rootDir: string) {
super();
}
// ── Database Access ────────────────────────────────────────────────
/**
* Get the SQLite database, initializing it on first access.
*/
private get db(): Database {
if (!this._db) {
const kbDir = `${this.rootDir}/.fusion`;
this._db = new Database(kbDir);
this._db.init();
}
return this._db;
}
/** Initialize the store (no-op, DB is lazily initialized). */
async init(): Promise<void> {
// Trigger lazy init
const _ = this.db;
}
// ── Row Conversion ─────────────────────────────────────────────────
private rowToRoutine(row: any): Routine {
const triggerConfig = fromJson<{
cronExpression?: string;
timezone?: string;
webhookPath?: string;
secret?: string;
endpoint?: string;
}>(row.triggerConfig);
let trigger: RoutineTrigger;
switch (row.triggerType as RoutineTriggerType) {
case "cron":
trigger = {
type: "cron",
cronExpression: triggerConfig?.cronExpression ?? "0 * * * *",
timezone: triggerConfig?.timezone,
} as RoutineCronTrigger;
break;
case "webhook":
trigger = {
type: "webhook",
webhookPath: triggerConfig?.webhookPath ?? "",
secret: triggerConfig?.secret,
} as RoutineWebhookTrigger;
break;
case "api":
trigger = {
type: "api",
endpoint: triggerConfig?.endpoint ?? "",
} as RoutineApiTrigger;
break;
case "manual":
default:
trigger = { type: "manual" } as RoutineManualTrigger;
break;
}
return {
id: row.id,
name: row.name,
description: row.description || undefined,
trigger,
catchUpPolicy: (row.catchUpPolicy as Routine["catchUpPolicy"]) || "run_one",
executionPolicy: (row.executionPolicy as Routine["executionPolicy"]) || "queue",
enabled: row.enabled === 1,
lastRunAt: row.lastRunAt || undefined,
lastRunResult: fromJson<RoutineExecutionResult>(row.lastRunResult),
nextRunAt: row.nextRunAt || undefined,
runCount: row.runCount || 0,
runHistory: fromJson<RoutineExecutionResult[]>(row.runHistory) || [],
cronExpression: isCronTrigger(trigger) ? trigger.cronExpression : undefined,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
};
}
private upsertRoutine(routine: Routine): void {
const trigger = routine.trigger;
let triggerConfig: Record<string, unknown> = {};
if (isCronTrigger(trigger)) {
triggerConfig = {
cronExpression: trigger.cronExpression,
timezone: trigger.timezone,
};
} else if (trigger.type === "webhook") {
triggerConfig = {
webhookPath: trigger.webhookPath,
secret: trigger.secret,
};
} else if (trigger.type === "api") {
triggerConfig = {
endpoint: trigger.endpoint,
};
}
this.db.prepare(`
INSERT OR REPLACE INTO routines (
id, name, description, triggerType, triggerConfig,
catchUpPolicy, executionPolicy, enabled,
lastRunAt, lastRunResult, nextRunAt,
runCount, runHistory, createdAt, updatedAt
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
routine.id,
routine.name,
routine.description ?? null,
trigger.type,
JSON.stringify(triggerConfig),
routine.catchUpPolicy,
routine.executionPolicy,
routine.enabled ? 1 : 0,
routine.lastRunAt ?? null,
routine.lastRunResult ? JSON.stringify(routine.lastRunResult) : null,
routine.nextRunAt ?? null,
routine.runCount || 0,
JSON.stringify(routine.runHistory || []),
routine.createdAt,
routine.updatedAt,
);
this.db.bumpLastModified();
}
// ── Locking ───────────────────────────────────────────────────────
/**
* Serialize all mutations to a given routine by chaining promises.
* Concurrent callers for the same ID will queue behind each other.
*/
private withRoutineLock<T>(id: string, fn: () => Promise<T>): Promise<T> {
const prev = this.routineLocks.get(id) ?? Promise.resolve();
let resolve!: () => void;
const next = new Promise<void>((r) => { resolve = r; });
this.routineLocks.set(id, next);
return prev.then(async () => {
try {
return await fn();
} finally {
if (this.routineLocks.get(id) === next) {
this.routineLocks.delete(id);
}
resolve!();
}
});
}
// ── Cron Utilities ────────────────────────────────────────────────
/**
* Compute the next run time from a cron expression.
* @param cronExpression - A valid cron expression (5 fields).
* @param fromDate - The date to compute from. Defaults to now.
* @returns ISO-8601 timestamp of the next run.
*/
computeNextRun(cronExpression: string, fromDate?: Date): string {
const interval = CronExpressionParser.parse(cronExpression, {
currentDate: fromDate ?? new Date(),
});
const next = interval.next();
return new Date(next.getTime()).toISOString();
}
/**
* Validate a cron expression. Returns true if valid.
*/
static isValidCron(cronExpression: string): boolean {
try {
CronExpressionParser.parse(cronExpression);
return true;
} catch {
return false;
}
}
// ── CRUD ──────────────────────────────────────────────────────────
/**
* Create a new routine.
*/
async createRoutine(input: RoutineCreateInput): Promise<Routine> {
if (!input.name?.trim()) {
throw new Error("Name is required and cannot be empty");
}
// Validate cron expression if cron trigger
if (isCronTrigger(input.trigger)) {
if (!RoutineStore.isValidCron(input.trigger.cronExpression)) {
throw new Error(`Invalid cron expression: "${input.trigger.cronExpression}"`);
}
}
const id = randomUUID();
const now = new Date().toISOString();
const enabled = input.enabled !== undefined ? input.enabled : true;
const routine: Routine = {
id,
name: input.name.trim(),
description: input.description?.trim() || undefined,
trigger: input.trigger,
catchUpPolicy: input.catchUpPolicy ?? "run_one",
executionPolicy: input.executionPolicy ?? "queue",
enabled,
runCount: 0,
runHistory: [],
createdAt: now,
updatedAt: now,
};
// Compute nextRunAt for enabled cron routines
if (enabled && isCronTrigger(routine.trigger)) {
routine.nextRunAt = this.computeNextRun(routine.trigger.cronExpression);
}
this.upsertRoutine(routine);
this.emit("routine:created", routine);
return routine;
}
/**
* Get a routine by ID.
*/
async getRoutine(id: string): Promise<Routine> {
const row = this.db.prepare("SELECT * FROM routines WHERE id = ?").get(id) as any;
if (!row) {
throw Object.assign(new Error(`Routine '${id}' not found`), { code: "ENOENT" });
}
return this.rowToRoutine(row);
}
/**
* List all routines.
*/
async listRoutines(): Promise<Routine[]> {
const rows = this.db.prepare("SELECT * FROM routines ORDER BY createdAt ASC").all() as any[];
return rows.map((row) => this.rowToRoutine(row));
}
/**
* Update an existing routine.
*/
async updateRoutine(id: string, updates: RoutineUpdateInput): Promise<Routine> {
return this.withRoutineLock(id, async () => {
const routine = await this.getRoutine(id);
if (updates.name !== undefined) {
if (!updates.name.trim()) throw new Error("Name cannot be empty");
routine.name = updates.name.trim();
}
if (updates.description !== undefined) {
routine.description = updates.description?.trim() || undefined;
}
if (updates.trigger !== undefined) {
// Validate cron if switching to cron
if (isCronTrigger(updates.trigger)) {
if (!RoutineStore.isValidCron(updates.trigger.cronExpression)) {
throw new Error(`Invalid cron expression: "${updates.trigger.cronExpression}"`);
}
}
routine.trigger = updates.trigger;
}
if (updates.catchUpPolicy !== undefined) {
routine.catchUpPolicy = updates.catchUpPolicy;
}
if (updates.executionPolicy !== undefined) {
routine.executionPolicy = updates.executionPolicy;
}
if (updates.enabled !== undefined) {
routine.enabled = updates.enabled;
}
// Recompute nextRunAt if enabled and cron trigger
if (routine.enabled && isCronTrigger(routine.trigger)) {
routine.nextRunAt = this.computeNextRun(routine.trigger.cronExpression);
} else if (!routine.enabled || !isCronTrigger(routine.trigger)) {
routine.nextRunAt = undefined;
}
routine.updatedAt = new Date().toISOString();
this.upsertRoutine(routine);
this.emit("routine:updated", routine);
return routine;
});
}
/**
* Delete a routine.
*/
async deleteRoutine(id: string): Promise<Routine> {
return this.withRoutineLock(id, async () => {
const routine = await this.getRoutine(id);
this.db.prepare("DELETE FROM routines WHERE id = ?").run(id);
this.db.bumpLastModified();
this.emit("routine:deleted", routine);
return routine;
});
}
// ── Run Tracking ─────────────────────────────────────────────────
/**
* Record a run result for a routine. Updates lastRunAt, lastRunResult,
* nextRunAt, runCount, and appends to runHistory.
*/
async recordRun(id: string, result: RoutineExecutionResult): Promise<Routine> {
return this.withRoutineLock(id, async () => {
const routine = await this.getRoutine(id);
routine.lastRunAt = result.startedAt;
routine.lastRunResult = result;
routine.runCount += 1;
// Prepend to history (most recent first), cap at MAX_ROUTINE_RUN_HISTORY
routine.runHistory.unshift(result);
if (routine.runHistory.length > MAX_ROUTINE_RUN_HISTORY) {
routine.runHistory = routine.runHistory.slice(0, MAX_ROUTINE_RUN_HISTORY);
}
// Recompute next run if enabled and cron trigger
if (routine.enabled && isCronTrigger(routine.trigger)) {
routine.nextRunAt = this.computeNextRun(routine.trigger.cronExpression);
}
routine.updatedAt = new Date().toISOString();
this.upsertRoutine(routine);
this.emit("routine:run", { routine, result });
return routine;
});
}
/**
* Get all routines that are due to run (nextRunAt <= now and enabled).
*/
async getDueRoutines(): Promise<Routine[]> {
const now = new Date().toISOString();
const rows = this.db.prepare(
"SELECT * FROM routines WHERE enabled = 1 AND nextRunAt IS NOT NULL AND nextRunAt <= ?"
).all(now) as any[];
return rows.map((row) => this.rowToRoutine(row));
}
}

View File

@@ -0,0 +1,177 @@
/**
* Routine domain types for first-class recurring task automation.
*
* Routines are similar to ScheduledTasks but support multiple trigger modes
* (cron, webhook, API, manual) with configurable execution and catch-up policies.
*/
import type { AutomationRunResult } from "./automation.js";
// ── Trigger Types ─────────────────────────────────────────────────────
/** Supported trigger modes for routines. */
export type RoutineTriggerType = "cron" | "webhook" | "api" | "manual";
/** Cron-based trigger with timezone support. */
export interface RoutineCronTrigger {
type: "cron";
/** Valid 5-field cron expression. */
cronExpression: string;
/** Optional IANA timezone (e.g., "America/New_York"). Defaults to UTC. */
timezone?: string;
}
/** Webhook trigger for external invocation. */
export interface RoutineWebhookTrigger {
type: "webhook";
/** URL path for the webhook (e.g., "/trigger/my-routine"). */
webhookPath: string;
/** Optional HMAC secret for signature verification. */
secret?: string;
}
/** API-triggered routine. */
export interface RoutineApiTrigger {
type: "api";
/** API endpoint that triggers this routine. */
endpoint: string;
}
/** Manually triggered routine. */
export interface RoutineManualTrigger {
type: "manual";
}
/** Union of all trigger types. */
export type RoutineTrigger =
| RoutineCronTrigger
| RoutineWebhookTrigger
| RoutineApiTrigger
| RoutineManualTrigger;
/** Discriminant helper for trigger type narrowing. */
export function isCronTrigger(trigger: RoutineTrigger): trigger is RoutineCronTrigger {
return trigger.type === "cron";
}
export function isWebhookTrigger(trigger: RoutineTrigger): trigger is RoutineWebhookTrigger {
return trigger.type === "webhook";
}
export function isApiTrigger(trigger: RoutineTrigger): trigger is RoutineApiTrigger {
return trigger.type === "api";
}
export function isManualTrigger(trigger: RoutineTrigger): trigger is RoutineManualTrigger {
return trigger.type === "manual";
}
// ── Execution Policies ─────────────────────────────────────────────────
/**
* Catch-up policy: what to do when a routine misses its scheduled run.
* - `run`: Execute the routine for each missed occurrence (catch-up runs).
* - `skip`: Skip missed occurrences entirely.
* - `run_one`: Execute once for the most recent missed occurrence only.
*/
export type RoutineCatchUpPolicy = "run" | "skip" | "run_one";
/**
* Execution policy: how to handle concurrent runs of the same routine.
* - `parallel`: Allow multiple concurrent executions.
* - `queue`: Queue subsequent runs, execute one at a time.
* - `reject`: Reject new runs if one is already in progress.
*/
export type RoutineExecutionPolicy = "parallel" | "queue" | "reject";
// ── Execution Result ───────────────────────────────────────────────────
/**
* Result of a single routine execution.
* Extends AutomationRunResult with routine-specific fields.
*/
export interface RoutineExecutionResult extends AutomationRunResult {
/** ID of the routine that was executed. */
routineId: string;
/** Whether a catch-up run was triggered. */
isCatchUp?: boolean;
/** Trigger type that fired this execution. */
triggerType?: RoutineTriggerType;
}
// ── Routine ───────────────────────────────────────────────────────────
/**
* A routine is a recurring automation with configurable triggers and policies.
*/
export interface Routine {
/** Unique identifier (UUID). */
id: string;
/** Human-readable name. */
name: string;
/** Optional description of what this routine does. */
description?: string;
/** The trigger configuration. */
trigger: RoutineTrigger;
/** Catch-up policy for missed runs. Default: "run_one". */
catchUpPolicy: RoutineCatchUpPolicy;
/** Execution policy for concurrent runs. Default: "queue". */
executionPolicy: RoutineExecutionPolicy;
/** Whether this routine is currently enabled. */
enabled: boolean;
/** ISO-8601 timestamp of the last run start, if any. */
lastRunAt?: string;
/** Result of the most recent run, if any. */
lastRunResult?: RoutineExecutionResult;
/** ISO-8601 timestamp of the next scheduled run (for cron triggers). */
nextRunAt?: string;
/** Total number of runs executed. */
runCount: number;
/** History of recent run results (most recent first, capped at MAX_ROUTINE_RUN_HISTORY). */
runHistory: RoutineExecutionResult[];
/** Optional cron expression stored directly for due-routine queries (derived from trigger). */
cronExpression?: string;
/** ISO-8601 timestamp of when this routine was created. */
createdAt: string;
/** ISO-8601 timestamp of when this routine was last updated. */
updatedAt: string;
}
// ── Input Types ───────────────────────────────────────────────────────
/** Input for creating a new routine. */
export interface RoutineCreateInput {
/** Human-readable name. Required. */
name: string;
/** Optional description. */
description?: string;
/** Trigger configuration. Required. */
trigger: RoutineTrigger;
/** Catch-up policy. Default: "run_one". */
catchUpPolicy?: RoutineCatchUpPolicy;
/** Execution policy. Default: "queue". */
executionPolicy?: RoutineExecutionPolicy;
/** Whether enabled. Default: true. */
enabled?: boolean;
}
/** Input for updating an existing routine. */
export interface RoutineUpdateInput {
/** Human-readable name. */
name?: string;
/** Optional description. */
description?: string;
/** Trigger configuration. */
trigger?: RoutineTrigger;
/** Catch-up policy. */
catchUpPolicy?: RoutineCatchUpPolicy;
/** Execution policy. */
executionPolicy?: RoutineExecutionPolicy;
/** Whether enabled. */
enabled?: boolean;
}
// ── Constants ─────────────────────────────────────────────────────────
/** Maximum number of run history entries to retain per routine. */
export const MAX_ROUTINE_RUN_HISTORY = 50;

View File

@@ -465,7 +465,7 @@ describe("Run Audit", () => {
});
it("schema version is bumped to 25", () => {
expect(db.getSchemaVersion()).toBe(26);
expect(db.getSchemaVersion()).toBe(27);
});
});
});