fusion/packages/core/src/routine-store.ts

/**
 * 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, 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";
import { assertProjectRootDir } from "./project-root-guard.js";
import type { AsyncDataLayer } from "./postgres/data-layer.js";
import { appendConfigurationRevision, createConfigurationRevision, getConfigurationRevision, rollbackConfiguration } from "./async-configuration-revision-store.js";
import type { ConfigChangedBy, ConfigurationRevision } from "./types.js";
/*
 * FNXC:SqliteFinalRemoval 2026-06-26-10:30:
 * Async Drizzle helpers for backend-mode (PostgreSQL) RoutineStore operations.
 * These helpers target the project.routines table via Drizzle and are the async
 * equivalent of the sync this.db.prepare() call sites below.
 */
import {
  upsertRoutine as upsertRoutineAsync,
  getRoutine as getRoutineAsync,
  listRoutines as listRoutinesAsync,
  deleteRoutine as deleteRoutineAsync,
  getDueRoutines as getDueRoutinesAsync,
} from "./async-routine-store.js";

const CRON_TIMEZONE = "UTC";

export interface RoutineStoreEvents {
  "routine:created": [routine: Routine];
  "routine:updated": [routine: Routine];
  "routine:deleted": [routine: Routine];
  "routine:run": [data: { routine: Routine; result: RoutineExecutionResult }];
}

/** Database row shape for the routines table. */
interface RoutineRow {
  id: string;
  agentId: string;
  name: string;
  description: string | null;
  triggerType: string;
  triggerConfig: string | null;
  command: string | null;
  steps: string | null;
  timeoutMs: number | null;
  catchUpPolicy: string;
  executionPolicy: string;
  enabled: number;
  lastRunAt: string | null;
  lastRunResult: string | null;
  nextRunAt: string | null;
  runCount: number;
  runHistory: string;
  catchUpLimit: number;
  scope: string;
  createdAt: string;
  updatedAt: string;
}

export interface RoutineStoreOptions {
  /**
   * FNXC:SqliteFinalRemoval 2026-06-26-10:30:
   * When an AsyncDataLayer is injected, RoutineStore operates in "backend mode":
   * all data access delegates to PostgreSQL via Drizzle and no SQLite Database
   * is constructed. When absent, the legacy SQLite path is byte-identical to
   * pre-migration. This mirrors the TaskStore/AgentStore dual-path pattern.
   */
  asyncLayer?: AsyncDataLayer;
}

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();

  /**
   * FNXC:SqliteFinalRemoval 2026-06-26-10:30:
   * When set, RoutineStore operates in backend mode (PostgreSQL via Drizzle).
   * All data access delegates to async helpers. No SQLite Database is
   * constructed. This mirrors the TaskStore/AgentStore dual-path pattern.
   */
  public readonly asyncLayer: AsyncDataLayer | null = null;

  /** True when AsyncDataLayer was injected. Gates all SQLite construction. */
  public get backendMode(): boolean {
    return this.asyncLayer !== null;
  }

  constructor(private rootDir: string, options?: RoutineStoreOptions) {
    super();
    assertProjectRootDir(rootDir, "RoutineStore");
    this.asyncLayer = options?.asyncLayer ?? null;
  }

  private requireVersionedConfigurationBackend(): void {
    /* FNXC:ConfigVersioning 2026-07-18-19:10: legacy SQLite routine writes have no durable atomic revision transaction, so reject before mutation rather than create non-rollbackable configuration. */
    if (!this.backendMode) throw new Error("Routine configuration changes require the PostgreSQL revision store");
  }

  // ── Database Access ────────────────────────────────────────────────

  /**
   * Get the SQLite database, initializing it on first access.
   *
   * FNXC:SqliteFinalRemoval 2026-06-26-10:30:
   * Throws in backend mode (asyncLayer injected) — callers must branch on
   * backendMode and use the async helpers instead.
   */
  private get db(): Database {
        throw new Error("SQLite Database is not available in backend mode (asyncLayer injected)");
}

  /** Initialize the store (no-op in backend mode, DB is lazily initialized otherwise). */
  async init(): Promise<void> {
    // FNXC:SqliteFinalRemoval 2026-06-26-10:30: No-op in backend mode.
        return;
}

  // ── Row Conversion ─────────────────────────────────────────────────

  private rowToRoutine(row: RoutineRow): 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,
      agentId: row.agentId || "",
      name: row.name,
      description: row.description || undefined,
      trigger,
      command: row.command || undefined,
      steps: fromJson<Routine["steps"]>(row.steps),
      timeoutMs: row.timeoutMs ?? undefined,
      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) || [],
      catchUpLimit: row.catchUpLimit ?? 5,
      cronExpression: isCronTrigger(trigger) ? trigger.cronExpression : undefined,
      scope: (row.scope as "global" | "project") || "project",
      createdAt: row.createdAt,
      updatedAt: row.updatedAt,
    };
  }

  private async upsertRoutine(routine: Routine): Promise<void> {
    /*
     * FNXC:SqliteFinalRemoval 2026-06-26-10:35:
     * Backend-mode: delegate to the async Drizzle upsertRoutine helper.
     */
        await upsertRoutineAsync(this.asyncLayer!.db, routine);
    return;
}

  // ── 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(),
      tz: CRON_TIMEZONE,
    });
    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 ──────────────────────────────────────────────────────────

  /*
   * FNXC:ConfigVersioning 2026-07-18-12:15:
   * Preserve the legacy SQLite CRUD seam while installations migrate. The
   * PostgreSQL branch below journals mutations atomically; compatibility
   * callers must not lose their pre-existing ability to manage routines.
   */

  /**
   * Create a new routine.
   */
  async createRoutine(input: RoutineCreateInput, changedBy: ConfigChangedBy = { kind: "human", id: "local-user" }): Promise<Routine> {
    this.requireVersionedConfigurationBackend();
    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,
      agentId: input.agentId,
      name: input.name.trim(),
      description: input.description?.trim() || undefined,
      trigger: input.trigger,
      command: input.command?.trim() || undefined,
      steps: input.steps && input.steps.length > 0 ? input.steps : undefined,
      timeoutMs: input.timeoutMs,
      catchUpPolicy: input.catchUpPolicy ?? "run_one",
      executionPolicy: input.executionPolicy ?? "queue",
      enabled,
      runCount: 0,
      runHistory: [],
      scope: input.scope ?? "project",
      createdAt: now,
      updatedAt: now,
    };

    // Compute nextRunAt for enabled cron routines
    if (enabled && isCronTrigger(routine.trigger)) {
      routine.nextRunAt = this.computeNextRun(routine.trigger.cronExpression);
    }

    /*
    FNXC:ConfigVersioning 2026-07-18-00:30:
    FN-8282 requires routine creation and its immutable snapshot to commit in
    one PostgreSQL transaction, so a failed revision insert cannot expose a
    routine with no restore point.
    */
        await this.asyncLayer!.transactionImmediate(async (tx) => {
      await upsertRoutineAsync(tx, routine);
      const revision = createConfigurationRevision({
        projectId: this.asyncLayer!.projectId ?? "",
        ownerScope: "project",
        configKind: "routine",
        configTarget: { routineId: routine.id },
        before: null,
        after: routine,
        changedBy,
      });
      if (revision) await appendConfigurationRevision(tx, revision);
    });

    this.emit("routine:created", routine);
    return routine;
  }

  /**
   * Get a routine by ID.
   */
  async getRoutine(id: string): Promise<Routine> {
    /*
     * FNXC:SqliteFinalRemoval 2026-06-26-10:40:
     * Backend-mode: delegate to the async Drizzle getRoutine helper.
     */
        return getRoutineAsync(this.asyncLayer!.db, id);
}

  /**
   * List all routines.
   */
  async listRoutines(): Promise<Routine[]> {
    /*
     * FNXC:SqliteFinalRemoval 2026-06-26-10:40:
     * Backend-mode: delegate to the async Drizzle listRoutines helper.
     */
        return listRoutinesAsync(this.asyncLayer!.db);
}

  /**
   * Update an existing routine.
   */
  async updateRoutine(id: string, updates: RoutineUpdateInput, changedBy: ConfigChangedBy = { kind: "human", id: "local-user" }): Promise<Routine> {
    this.requireVersionedConfigurationBackend();
    return this.withRoutineLock(id, async () => {
      const routine = await this.getRoutine(id);
      const before = structuredClone(routine);

      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.command !== undefined) {
        routine.command = updates.command?.trim() || undefined;
      }
      if (updates.steps !== undefined) {
        routine.steps = updates.steps.length > 0 ? updates.steps : undefined;
      }
      if (updates.timeoutMs !== undefined) {
        routine.timeoutMs = updates.timeoutMs;
      }
      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();
            await this.asyncLayer!.transactionImmediate(async (tx) => {
        await upsertRoutineAsync(tx, routine);
        const revision = createConfigurationRevision({
          projectId: this.asyncLayer!.projectId ?? "",
          ownerScope: "project",
          configKind: "routine",
          configTarget: { routineId: routine.id },
          before,
          after: routine,
          changedBy,
        });
        if (revision) await appendConfigurationRevision(tx, revision);
      });

      this.emit("routine:updated", routine);
      return routine;
    });
  }

  /**
   * Restore a routine snapshot by stable id, including deletion when the
   * selected revision predates creation. The replacement and forward revision
   * share one backend transaction.
   */
  async rollbackConfiguration(revisionId: string, changedBy: ConfigChangedBy = { kind: "human", id: "local-user" }): Promise<ConfigurationRevision> {
    if (!this.backendMode) throw new Error("Configuration rollback requires the PostgreSQL revision store");
    const layer = this.asyncLayer!;
    return layer.transactionImmediate((tx) => rollbackConfiguration(tx, layer.projectId ?? "", revisionId, changedBy, {
      readCurrent: async () => {
        const revision = await getConfigurationRevision(tx, layer.projectId ?? "", revisionId);
        if (!revision || revision.configKind !== "routine") throw new Error(`Routine configuration revision ${revisionId} was not found`);
        try { return await getRoutineAsync(tx, String(revision.configTarget.routineId)); } catch (error) { if ((error as { code?: string }).code === "ENOENT") return null; throw error; }
      },
      replace: async (snapshot) => {
        const target = await getConfigurationRevision(tx, layer.projectId ?? "", revisionId);
        const id = String(target?.configTarget.routineId ?? "");
        if (snapshot === null) await deleteRoutineAsync(tx, id);
        else await upsertRoutineAsync(tx, snapshot as Routine);
      },
    }));
  }

  /**
   * Delete a routine.
   */
  async deleteRoutine(id: string, changedBy: ConfigChangedBy = { kind: "human", id: "local-user" }): Promise<Routine> {
    this.requireVersionedConfigurationBackend();
    return this.withRoutineLock(id, async () => {
      const routine = await this.getRoutine(id);
      /*
       * FNXC:SqliteFinalRemoval 2026-06-26-10:40:
       * Backend-mode: delegate to the async Drizzle deleteRoutine helper.
       */
            await this.asyncLayer!.transactionImmediate(async (tx) => {
        await deleteRoutineAsync(tx, id);
        const revision = createConfigurationRevision({
          projectId: this.asyncLayer!.projectId ?? "",
          ownerScope: "project",
          configKind: "routine",
          configTarget: { routineId: routine.id },
          before: routine,
          after: null,
          changedBy,
        });
        if (revision) await appendConfigurationRevision(tx, revision);
      });

      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();
      await this.upsertRoutine(routine);
      this.emit("routine:run", { routine, result });
      return routine;
    });
  }

  /**
   * Mark a routine execution as started (pre-run bookkeeping).
   */
  async startRoutineExecution(
    id: string,
    meta: { triggeredAt: string; catchUpFrom?: string; invocationSource: string },
  ): Promise<void> {
    await this.withRoutineLock(id, async () => {
      const routine = await this.getRoutine(id);
      routine.lastRunAt = meta.triggeredAt;
      routine.updatedAt = new Date().toISOString();
      await this.upsertRoutine(routine);
    });
  }

  /**
   * Record the completion (success or failure) of a routine execution.
   */
  async completeRoutineExecution(
    id: string,
    meta: { completedAt: string; success: boolean; resultJson?: Record<string, unknown>; error?: string; output?: string; triggerType?: RoutineTriggerType; stepResults?: RoutineExecutionResult["stepResults"] },
  ): Promise<void> {
    await this.withRoutineLock(id, async () => {
      const routine = await this.getRoutine(id);
      const result: RoutineExecutionResult = {
        routineId: id,
        success: meta.success,
        output: meta.output ?? (meta.success ? JSON.stringify(meta.resultJson ?? {}) : ""),
        error: meta.error,
        startedAt: routine.lastRunAt ?? meta.completedAt,
        completedAt: meta.completedAt,
        triggerType: meta.triggerType,
        stepResults: meta.stepResults,
      };

      routine.lastRunAt = result.startedAt;
      routine.lastRunResult = result;
      routine.runCount += 1;

      routine.runHistory.unshift(result);
      if (routine.runHistory.length > MAX_ROUTINE_RUN_HISTORY) {
        routine.runHistory = routine.runHistory.slice(0, MAX_ROUTINE_RUN_HISTORY);
      }

      if (routine.enabled && isCronTrigger(routine.trigger)) {
        routine.nextRunAt = this.computeNextRun(routine.trigger.cronExpression);
      }

      routine.updatedAt = new Date().toISOString();
      await this.upsertRoutine(routine);
      this.emit("routine:run", { routine, result });
    });
  }

  /**
   * Cancel a routine execution (no result recorded, just reset state).
   */
  async cancelRoutineExecution(id: string): Promise<void> {
    await this.withRoutineLock(id, async () => {
      const routine = await this.getRoutine(id);
      if (routine.enabled && isCronTrigger(routine.trigger)) {
        routine.nextRunAt = this.computeNextRun(routine.trigger.cronExpression);
      }
      routine.updatedAt = new Date().toISOString();
      await this.upsertRoutine(routine);
    });
  }

  /**
   * Get all routines that are due to run (nextRunAt <= now and enabled).
   * Filters by scope: "global" or "project".
   */
  async getDueRoutines(scope: "global" | "project"): Promise<Routine[]> {
    const now = new Date().toISOString();
    /*
     * FNXC:SqliteFinalRemoval 2026-06-26-10:45:
     * Backend-mode: delegate to the async Drizzle getDueRoutines helper.
     */
        return getDueRoutinesAsync(this.asyncLayer!.db, now, scope);
}

  /**
   * Get all routines that are due to run (nextRunAt <= now and enabled) for both scopes.
   * Returns routines from both "global" and "project" scopes.
   */
  async getDueRoutinesAllScopes(): Promise<Routine[]> {
    const now = new Date().toISOString();
    /*
     * FNXC:SqliteFinalRemoval 2026-06-26-10:45:
     * Backend-mode: delegate to the async Drizzle getDueRoutines helper (no scope).
     */
        return getDueRoutinesAsync(this.asyncLayer!.db, now);
}
}