feat(FN-2147): merge fusion/fn-2147

This commit is contained in:
gsxdsm
2026-04-19 05:37:09 -07:00
parent 727b0d1f56
commit 86aa4c19f7
2 changed files with 26 additions and 68 deletions

View File

@@ -2,7 +2,7 @@ import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { AutomationStore } from "./automation-store.js"; import { AutomationStore } from "./automation-store.js";
import { rm } from "node:fs/promises"; import { rm } from "node:fs/promises";
import { join } from "node:path"; import { join } from "node:path";
import { mkdtempSync, existsSync } from "node:fs"; import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os"; import { tmpdir } from "node:os";
import type { ScheduledTask, AutomationRunResult, AutomationStep } from "./automation.js"; import type { ScheduledTask, AutomationRunResult, AutomationStep } from "./automation.js";
import { randomUUID } from "node:crypto"; import { randomUUID } from "node:crypto";
@@ -39,16 +39,13 @@ describe("AutomationStore", () => {
// ── init ────────────────────────────────────────────────────────── // ── init ──────────────────────────────────────────────────────────
describe("init", () => { describe("init", () => {
it("creates the automations directory", async () => { it("initializes database-backed store", async () => {
const dir = join(rootDir, ".fusion", "automations"); await expect(store.init()).resolves.toBeUndefined();
expect(existsSync(dir)).toBe(true);
}); });
it("is idempotent", async () => { it("is idempotent", async () => {
await store.init(); await expect(store.init()).resolves.toBeUndefined();
await store.init(); await expect(store.init()).resolves.toBeUndefined();
const dir = join(rootDir, ".fusion", "automations");
expect(existsSync(dir)).toBe(true);
}); });
}); });
@@ -175,15 +172,20 @@ describe("AutomationStore", () => {
).rejects.toThrow("Invalid cron expression"); ).rejects.toThrow("Invalid cron expression");
}); });
it("persists schedule to disk", async () => { it("persists schedule to database", async () => {
const schedule = await store.createSchedule({ const schedule = await store.createSchedule({
name: "Persist test", name: "Persist test",
command: "echo persist", command: "echo persist",
scheduleType: "weekly", scheduleType: "weekly",
}); });
const filePath = join(rootDir, ".fusion", "automations", `${schedule.id}.json`); const secondStore = new AutomationStore(rootDir);
expect(existsSync(filePath)).toBe(true); await secondStore.init();
const reloaded = await secondStore.getSchedule(schedule.id);
expect(reloaded.id).toBe(schedule.id);
expect(reloaded.name).toBe("Persist test");
expect(reloaded.cronExpression).toBe("0 0 * * 1");
}); });
it("emits schedule:created event", async () => { it("emits schedule:created event", async () => {
@@ -364,8 +366,7 @@ describe("AutomationStore", () => {
const deleted = await store.deleteSchedule(schedule.id); const deleted = await store.deleteSchedule(schedule.id);
expect(deleted.id).toBe(schedule.id); expect(deleted.id).toBe(schedule.id);
const filePath = join(rootDir, ".fusion", "automations", `${schedule.id}.json`); await expect(store.getSchedule(schedule.id)).rejects.toThrow("not found");
expect(existsSync(filePath)).toBe(false);
}); });
it("throws for missing schedule", async () => { it("throws for missing schedule", async () => {

View File

@@ -1,7 +1,5 @@
import { EventEmitter } from "node:events"; import { EventEmitter } from "node:events";
import { mkdir, readFile, readdir, rename, writeFile } from "node:fs/promises";
import { join } from "node:path"; import { join } from "node:path";
import { existsSync } from "node:fs";
import { randomUUID } from "node:crypto"; import { randomUUID } from "node:crypto";
import { CronExpressionParser } from "cron-parser"; import { CronExpressionParser } from "cron-parser";
import type { import type {
@@ -22,7 +20,6 @@ export interface AutomationStoreEvents {
} }
export class AutomationStore extends EventEmitter<AutomationStoreEvents> { export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
private automationsDir: string;
/** Per-schedule promise chain for serializing writes. */ /** Per-schedule promise chain for serializing writes. */
private scheduleLocks: Map<string, Promise<void>> = new Map(); private scheduleLocks: Map<string, Promise<void>> = new Map();
/** SQLite database instance */ /** SQLite database instance */
@@ -30,7 +27,6 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
constructor(private rootDir: string) { constructor(private rootDir: string) {
super(); super();
this.automationsDir = join(rootDir, ".fusion", "automations");
} }
/** /**
@@ -49,8 +45,6 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
async init(): Promise<void> { async init(): Promise<void> {
// Ensure DB is initialized // Ensure DB is initialized
const _ = this.db; const _ = this.db;
// Keep automations dir for backward compat
await mkdir(this.automationsDir, { recursive: true });
} }
// ── Row Conversion ───────────────────────────────────────────────── // ── Row Conversion ─────────────────────────────────────────────────
@@ -103,13 +97,12 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
schedule.createdAt, schedule.createdAt,
schedule.updatedAt, schedule.updatedAt,
); );
this.db.bumpLastModified();
} }
// ── Locking ──────────────────────────────────────────────────────── // ── Locking ────────────────────────────────────────────────────────
/** /**
* Serialize all mutations to a given schedule's JSON file by chaining promises. * Serialize all mutations to a given schedule by chaining promises.
* Concurrent callers for the same ID will queue behind each other. * Concurrent callers for the same ID will queue behind each other.
*/ */
private withScheduleLock<T>(id: string, fn: () => Promise<T>): Promise<T> { private withScheduleLock<T>(id: string, fn: () => Promise<T>): Promise<T> {
@@ -130,43 +123,19 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
}); });
} }
// ── File I/O ─────────────────────────────────────────────────────── // ── Persistence ────────────────────────────────────────────────────
private schedulePath(id: string): string {
return join(this.automationsDir, `${id}.json`);
}
private async readScheduleJson(id: string): Promise<ScheduledTask> { private async readScheduleJson(id: string): Promise<ScheduledTask> {
// Read from SQLite first
const row = this.db.prepare('SELECT * FROM automations WHERE id = ?').get(id); const row = this.db.prepare('SELECT * FROM automations WHERE id = ?').get(id);
if (row) return this.rowToSchedule(row); if (!row) {
throw Object.assign(new Error(`Schedule '${id}' not found`), { code: "ENOENT" });
// Fallback to file
const filePath = this.schedulePath(id);
const raw = await readFile(filePath, "utf-8");
try {
return JSON.parse(raw) as ScheduledTask;
} catch (err) {
throw new Error(
`Failed to parse schedule JSON at ${filePath}: ${(err as Error).message}`,
);
} }
return this.rowToSchedule(row);
} }
/** private async persistSchedule(schedule: ScheduledTask): Promise<void> {
* Write a schedule to SQLite and also to disk for backward compat.
*/
private async atomicWriteScheduleJson(id: string, schedule: ScheduledTask): Promise<void> {
this.upsertSchedule(schedule); this.upsertSchedule(schedule);
// Also write to disk for backward compatibility this.db.bumpLastModified();
try {
const filePath = this.schedulePath(id);
const tmpPath = filePath + ".tmp";
await writeFile(tmpPath, JSON.stringify(schedule, null, 2));
await rename(tmpPath, filePath);
} catch {
// Non-fatal
}
} }
// ── Cron Computation ─────────────────────────────────────────────── // ── Cron Computation ───────────────────────────────────────────────
@@ -244,17 +213,13 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
updatedAt: now, updatedAt: now,
}; };
await this.atomicWriteScheduleJson(id, schedule); await this.persistSchedule(schedule);
this.emit("schedule:created", schedule); this.emit("schedule:created", schedule);
return schedule; return schedule;
} }
async getSchedule(id: string): Promise<ScheduledTask> { async getSchedule(id: string): Promise<ScheduledTask> {
const row = this.db.prepare('SELECT * FROM automations WHERE id = ?').get(id); return this.readScheduleJson(id);
if (!row) {
throw Object.assign(new Error(`Schedule '${id}' not found`), { code: "ENOENT" });
}
return this.rowToSchedule(row);
} }
async listSchedules(): Promise<ScheduledTask[]> { async listSchedules(): Promise<ScheduledTask[]> {
@@ -318,7 +283,7 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
} }
schedule.updatedAt = new Date().toISOString(); schedule.updatedAt = new Date().toISOString();
await this.atomicWriteScheduleJson(id, schedule); await this.persistSchedule(schedule);
this.emit("schedule:updated", schedule); this.emit("schedule:updated", schedule);
return schedule; return schedule;
}); });
@@ -352,7 +317,7 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
schedule.steps = reordered; schedule.steps = reordered;
schedule.updatedAt = new Date().toISOString(); schedule.updatedAt = new Date().toISOString();
await this.atomicWriteScheduleJson(scheduleId, schedule); await this.persistSchedule(schedule);
this.emit("schedule:updated", schedule); this.emit("schedule:updated", schedule);
return schedule; return schedule;
}); });
@@ -364,14 +329,6 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
// Delete from SQLite // Delete from SQLite
this.db.prepare('DELETE FROM automations WHERE id = ?').run(id); this.db.prepare('DELETE FROM automations WHERE id = ?').run(id);
this.db.bumpLastModified(); this.db.bumpLastModified();
// Also remove file for backward compat
try {
const filePath = this.schedulePath(id);
const { unlink } = await import("node:fs/promises");
await unlink(filePath);
} catch {
// Non-fatal
}
this.emit("schedule:deleted", schedule); this.emit("schedule:deleted", schedule);
return schedule; return schedule;
}); });
@@ -401,7 +358,7 @@ export class AutomationStore extends EventEmitter<AutomationStoreEvents> {
} }
schedule.updatedAt = new Date().toISOString(); schedule.updatedAt = new Date().toISOString();
await this.atomicWriteScheduleJson(id, schedule); await this.persistSchedule(schedule);
this.emit("schedule:run", { schedule, result }); this.emit("schedule:run", { schedule, result });
return schedule; return schedule;
}); });