feat(automations): add per-host maintenance mode and recurring schedules (#1521)

plugins/automations/src/backend/index.ts

@@ -9,6 +9,9 @@ import { registerRoutes } from "./routes.js";
 import { createScheduler, STARTUP_DELAY_MS, TICK_MS } from "./scheduler.js";
 import { createAutomationsService } from "./service.js";
 import { createTriggers } from "./triggers.js";
+import { createMaintenanceRepository } from "./maintenance-repository.js";
+import { createMaintenanceService } from "./maintenance-service.js";
+import { registerMaintenanceRoutes } from "./maintenance-routes.js";
 
 export type { AutomationsAccessV1, AutomationSummary } from "./service.js";
 
@@ -17,9 +20,30 @@ export async function activate(ctx: PluginContext) {
     const channels = await ctx.notify.channels();
     return channels.map((channel) => channel.id);
   });
+  const maintenance = createMaintenanceService(
+    ctx,
+    await createMaintenanceRepository(ctx.db),
+    async (userId, hostId) => {
+      for (const state of await repository.listOpenBreaches()) {
+        if (Number(state.stateKey.split(":")[0]) !== hostId) continue;
+        const automation = await repository.findById(state.automationId);
+        if (automation?.userId === userId)
+          await repository.clearBreach(state.automationId, state.stateKey);
+      }
+    },
+  );
+  ctx.events.on("user.data_wiped", (payload) => {
+    const userId = (payload as { userId?: string } | undefined)?.userId;
+    if (userId) return maintenance.wipeUser(userId);
+  });
   const deps = createDeps(ctx.services);
   const engine = new AutomationEngine(ctx, repository, deps);
-  const triggers = createTriggers(repository, engine, ctx.log);
+  const triggers = createTriggers(
+    repository,
+    engine,
+    ctx.log,
+    maintenance.isMaintaining,
+  );
   triggers.subscribe(ctx.events);
 
   const viewers = createHeadlessViewers(ctx, deps, () =>
@@ -40,15 +64,22 @@ export async function activate(ctx: PluginContext) {
     repository,
     engine,
     log: ctx.log,
-    reconcile: [() => viewers.reconcile(), (now) => docker.reconcile(now)],
+    reconcile: [
+      () => maintenance.tick(),
+      () => viewers.reconcile(),
+      (now) => docker.reconcile(now),
+    ],
+    isMaintaining: maintenance.isMaintaining,
   });
   ctx.schedule.after(STARTUP_DELAY_MS, () => scheduler.tick());
   ctx.schedule.every(TICK_MS, () => scheduler.tick());
 
+  const router = ctx.http.router<Router>({ public: ["/webhook/:token"] });
+  registerMaintenanceRoutes(router, ctx, maintenance);
   registerRoutes(
     // The webhook route authenticates on its own per-automation token rather
     // than a session, which is the point of an inbound webhook.
-    ctx.http.router<Router>({ public: ["/webhook/:token"] }),
+    router,
     { ctx, repository, engine, deps },
   );
 

plugins/automations/src/backend/maintenance-repository.ts

@@ -0,0 +1,66 @@
+import { and, eq } from "drizzle-orm";
+import type { PluginDatabase } from "@termix/plugin-sdk/backend";
+import { maintenanceTable } from "./tables.js";
+import { emptyMaintenance, type HostMaintenance } from "../maintenance.js";
+
+export interface MaintenanceRow {
+  userId: string;
+  hostId: number;
+  state: HostMaintenance;
+}
+export type MaintenanceRepository = Awaited<
+  ReturnType<typeof createMaintenanceRepository>
+>;
+
+export async function createMaintenanceRepository(db: PluginDatabase) {
+  // The SDK constructs dialect-specific tables at runtime.
+  /* eslint-disable @typescript-eslint/no-explicit-any */
+  const table: any = await db.define(maintenanceTable);
+  const client = () => db.client<any>();
+  /* eslint-enable @typescript-eslint/no-explicit-any */
+  const where = (userId: string, hostId: number) =>
+    and(eq(table.userId, userId), eq(table.hostId, hostId));
+  return {
+    async read(userId: string, hostId: number): Promise<HostMaintenance> {
+      const rows = await (
+        await client()
+      )
+        .select()
+        .from(table)
+        .where(where(userId, hostId));
+      return rows.length ? JSON.parse(rows[0].state) : emptyMaintenance();
+    },
+    async write(
+      userId: string,
+      hostId: number,
+      state: HostMaintenance,
+    ): Promise<void> {
+      const connection = await client();
+      const rows = await connection
+        .select({ id: table.id })
+        .from(table)
+        .where(where(userId, hostId));
+      const values = { state: JSON.stringify(state) };
+      if (rows.length)
+        await connection.update(table).set(values).where(where(userId, hostId));
+      else await connection.insert(table).values({ userId, hostId, ...values });
+      await db.persist();
+    },
+    async list(userId?: string): Promise<MaintenanceRow[]> {
+      const query = (await client()).select().from(table);
+      const rows = await (userId
+        ? query.where(eq(table.userId, userId))
+        : query);
+      return rows.map(
+        (row: { userId: string; hostId: number; state: string }) => ({
+          ...row,
+          state: JSON.parse(row.state),
+        }),
+      );
+    },
+    async wipeUser(userId: string): Promise<void> {
+      await (await client()).delete(table).where(eq(table.userId, userId));
+      await db.persist();
+    },
+  };
+}

plugins/automations/src/backend/maintenance-routes.ts

@@ -0,0 +1,78 @@
+import type { Router, Request, Response } from "express";
+import type { PluginContext } from "@termix/plugin-sdk/backend";
+import { MaintenanceInputError } from "./maintenance-time.js";
+import type { MaintenanceService } from "./maintenance-service.js";
+
+export function registerMaintenanceRoutes(
+  router: Router,
+  ctx: PluginContext,
+  service: MaintenanceService,
+) {
+  router.get(
+    "/maintenance",
+    ctx.rbac.require("view") as never,
+    async (_req: Request, res: Response) => {
+      try {
+        const user = ctx.currentActor();
+        if (!user)
+          return res.status(401).json({ error: "Authentication required" });
+        const hosts = new Set(
+          (await ctx.hosts.list())
+            .filter((host) => host.userId === user)
+            .map((host) => host.id),
+        );
+        res.json(
+          (await service.list(user)).filter((row) => hosts.has(row.hostId)),
+        );
+      } catch (error) {
+        ctx.log.error("Failed to list maintenance", error as Error);
+        res.status(500).json({ error: "Failed to list maintenance" });
+      }
+    },
+  );
+  const handle = (mutate: boolean) => async (req: Request, res: Response) => {
+    const hostId = Number(req.params.hostId);
+    if (!Number.isSafeInteger(hostId) || hostId < 1)
+      return res.status(400).json({ error: "Invalid host id" });
+    const user = ctx.currentActor();
+    if (!user)
+      return res.status(401).json({ error: "Authentication required" });
+    try {
+      const host = await ctx.hosts.get(hostId);
+      if (!host || host.userId !== user)
+        return res.status(404).json({ error: "Host not found" });
+      if (!mutate) return res.json(await service.read(user, hostId));
+      const state = await service.edit(
+        user,
+        hostId,
+        req.body?.action,
+        req.body,
+      );
+      await ctx.audit.record({
+        action: `maintenance_${req.body.action}`,
+        resourceType: "host",
+        resourceId: String(hostId),
+        resourceName: host.name || host.ip,
+        success: true,
+      });
+      res.json(state);
+    } catch (error) {
+      // Validation failures have no side effects; storage failures remain errors.
+      if (error instanceof MaintenanceInputError) {
+        return res.status(400).json({ error: error.message });
+      }
+      ctx.log.error("Failed to update maintenance", error as Error);
+      res.status(500).json({ error: "Failed to update maintenance" });
+    }
+  };
+  router.get(
+    "/maintenance/:hostId",
+    ctx.rbac.require("view") as never,
+    handle(false),
+  );
+  router.post(
+    "/maintenance/:hostId",
+    ctx.rbac.require("edit") as never,
+    handle(true),
+  );
+}

plugins/automations/src/backend/maintenance-service.ts

@@ -0,0 +1,152 @@
+import { randomUUID } from "node:crypto";
+import type { PluginContext } from "@termix/plugin-sdk/backend";
+import type { HostMaintenance } from "../maintenance.js";
+import type { MaintenanceRepository } from "./maintenance-repository.js";
+import {
+  MaintenanceInputError,
+  advanceMaintenance,
+  beginMaintenance,
+  parsePlan,
+  plansOverlap,
+} from "./maintenance-time.js";
+
+export type MaintenanceService = ReturnType<typeof createMaintenanceService>;
+
+export function createMaintenanceService(
+  ctx: PluginContext,
+  repository: MaintenanceRepository,
+  resetBreaches: (
+    userId: string,
+    hostId: number,
+  ) => Promise<void> = async () => {},
+) {
+  // Serialize reads that advance schedules with edits and the reminder tick.
+  let tail: Promise<unknown> = Promise.resolve();
+  const exclusive = <T>(work: () => Promise<T>): Promise<T> => {
+    const result = tail.then(work);
+    tail = result.catch(() => undefined);
+    return result;
+  };
+  async function update(
+    userId: string,
+    hostId: number,
+    edit?: (state: HostMaintenance) => void,
+  ) {
+    const state = await repository.read(userId, hostId);
+    const previous = JSON.stringify(state);
+    const wasActive = !!state.active;
+    advanceMaintenance(state, Date.now());
+    const activated = !wasActive && !!state.active;
+    edit?.(state);
+    if (activated || wasActive !== !!state.active)
+      await resetBreaches(userId, hostId);
+    if (JSON.stringify(state) !== previous)
+      await repository.write(userId, hostId, state);
+    return state;
+  }
+  const service = {
+    read: (userId: string, hostId: number) =>
+      exclusive(() => update(userId, hostId)),
+    async isMaintaining(userId: string, hostId: number): Promise<boolean> {
+      return !!(await service.read(userId, hostId)).active;
+    },
+    edit: (userId: string, hostId: number, action: string, input: unknown) =>
+      exclusive(() =>
+        update(userId, hostId, (state) => {
+          if (action === "end") {
+            state.active = null;
+            return;
+          }
+          if (action === "remove") {
+            const id = (input as { id?: unknown } | null)?.id;
+            if (
+              typeof id !== "string" ||
+              !state.plans.some((plan) => plan.id === id)
+            )
+              throw new MaintenanceInputError("Schedule not found");
+            state.plans = state.plans.filter((plan) => plan.id !== id);
+            return;
+          }
+          if (action !== "start" && action !== "schedule")
+            throw new MaintenanceInputError("Invalid maintenance action");
+          if (action === "start" && state.active)
+            throw new MaintenanceInputError("Maintenance is already active");
+          const plan = parsePlan(input, Date.now(), action === "start");
+          plan.id = randomUUID();
+          if (action === "start") {
+            beginMaintenance(state, plan, Date.now());
+            return;
+          }
+          if (state.plans.length >= 20)
+            throw new MaintenanceInputError(
+              "At most 20 maintenance schedules are allowed per host",
+            );
+          if (
+            state.plans.some(
+              (other) =>
+                other.nextStart &&
+                plansOverlap(plan, { ...other, start: other.nextStart }),
+            )
+          ) {
+            throw new MaintenanceInputError(
+              "Planned maintenance times overlap",
+            );
+          }
+          state.plans.push(plan);
+        }),
+      ),
+    async list(userId: string) {
+      const rows = await repository.list(userId);
+      return Promise.all(
+        rows.map(async (row) => ({
+          hostId: row.hostId,
+          state: await service.read(userId, row.hostId),
+        })),
+      );
+    },
+    wipeUser: (userId: string) => exclusive(() => repository.wipeUser(userId)),
+    async tick() {
+      for (const row of await repository.list()) {
+        try {
+          await exclusive(async () => {
+            const state = await update(row.userId, row.hostId);
+            const active = state.active;
+            if (!active?.notifyOverdue || active.notified) return;
+            const deadline =
+              Date.parse(active.estimatedEnd) + active.graceMinutes * 60_000;
+            if (Date.now() < deadline) return;
+            await ctx.asUser(row.userId, async () => {
+              const host = await ctx.hosts.get(row.hostId);
+              if (!host || host.userId !== row.userId) return;
+              const status = await ctx.hosts.status.get(row.hostId);
+              // Unknown or stale data is not evidence of an overdue outage.
+              if (
+                status?.status !== "offline" ||
+                Date.parse(status.lastChecked) < deadline ||
+                !Number.isFinite(Date.parse(status.lastChecked))
+              )
+                return;
+              const result = await ctx.notify.send({
+                title: `Maintenance overdue: ${host.name || host.ip}`,
+                body: `${active.reasons.join("; ")}. Estimated completion: ${active.estimatedEnd}. The host is still unavailable after the grace period. Maintenance remains active until you end it.`,
+                severity: "warning",
+                category: "automations.maintenance_overdue",
+                dedupeKey: `maintenance:${row.hostId}:${active.startedAt}:${active.estimatedEnd}`,
+                context: { hostId: row.hostId },
+              });
+              if (result.recipients > 0) {
+                active.notified = true;
+                await repository.write(row.userId, row.hostId, state);
+              }
+            });
+          });
+        } catch (error) {
+          ctx.log.warn(
+            `Maintenance check for host ${row.hostId} failed: ${String(error)}`,
+          );
+        }
+      }
+    },
+  };
+  return service;
+}

plugins/automations/src/backend/maintenance-time.ts

@@ -0,0 +1,155 @@
+import type { MaintenancePlan, HostMaintenance } from "../maintenance.js";
+
+export class MaintenanceInputError extends Error {}
+
+const DAY = 86_400_000;
+const WEEK = 7 * DAY;
+
+/** UTC wall-clock recurrence; monthly dates missing from a month are skipped. */
+export function nextOccurrence(
+  plan: MaintenancePlan,
+  after: number,
+): number | null {
+  const first = Date.parse(plan.start);
+  if (first > after) return first;
+  if (plan.recurrence === "once") return null;
+  if (plan.recurrence === "weekly") {
+    return first + (Math.floor((after - first) / WEEK) + 1) * WEEK;
+  }
+  const start = new Date(first);
+  const from = new Date(after);
+  let month = from.getUTCFullYear() * 12 + from.getUTCMonth();
+  for (let i = 0; i < 14; i++, month++) {
+    const candidate = Date.UTC(
+      Math.floor(month / 12),
+      month % 12,
+      start.getUTCDate(),
+      start.getUTCHours(),
+      start.getUTCMinutes(),
+    );
+    if (new Date(candidate).getUTCMonth() === month % 12 && candidate > after)
+      return candidate;
+  }
+  throw new MaintenanceInputError("Cannot calculate maintenance recurrence");
+}
+
+/** Exhaust a Gregorian cycle for recurring pairs, not an arbitrary lookahead. */
+export function plansOverlap(a: MaintenancePlan, b: MaintenancePlan): boolean {
+  const start = Math.max(Date.parse(a.start), Date.parse(b.start));
+  const end =
+    a.recurrence === "once" || b.recurrence === "once"
+      ? start + WEEK
+      : start + 146097 * DAY + WEEK;
+  let left = nextOccurrence(a, start - WEEK - 1);
+  let right = nextOccurrence(b, start - WEEK - 1);
+  while (left !== null && right !== null && left <= end && right <= end) {
+    const leftEnd = left + a.durationMinutes * 60_000;
+    const rightEnd = right + b.durationMinutes * 60_000;
+    if (left < rightEnd && right < leftEnd) return true;
+    if (leftEnd <= right) left = nextOccurrence(a, left);
+    else right = nextOccurrence(b, right);
+  }
+  return false;
+}
+
+export function beginMaintenance(
+  state: HostMaintenance,
+  plan: MaintenancePlan,
+  start: number,
+): void {
+  const end = new Date(start + plan.durationMinutes * 60_000).toISOString();
+  const current = state.active;
+  state.active = {
+    startedAt: new Date(
+      Math.min(start, current ? Date.parse(current.startedAt) : start),
+    ).toISOString(),
+    reasons: [...new Set([...(current?.reasons ?? []), plan.reason])].slice(
+      -20,
+    ),
+    estimatedEnd:
+      current && current.estimatedEnd > end ? current.estimatedEnd : end,
+    graceMinutes: Math.max(current?.graceMinutes ?? 0, plan.graceMinutes),
+    notifyOverdue: !!current?.notifyOverdue || plan.notifyOverdue,
+    notified: current && current.estimatedEnd >= end ? current.notified : false,
+  };
+}
+
+/** Catch up after downtime, preserving maintenance until explicitly ended. */
+export function advanceMaintenance(state: HostMaintenance, now: number): void {
+  for (const plan of state.plans) {
+    if (!plan.nextStart || Date.parse(plan.nextStart) > now) continue;
+    let due = Date.parse(plan.nextStart);
+    let next = nextOccurrence(plan, due);
+    while (next !== null && next <= now) {
+      due = next;
+      next = nextOccurrence(plan, due);
+    }
+    beginMaintenance(state, plan, due);
+    plan.nextStart = next === null ? null : new Date(next).toISOString();
+  }
+}
+
+export function parsePlan(
+  input: unknown,
+  now: number,
+  immediate = false,
+): MaintenancePlan {
+  if (!input || typeof input !== "object")
+    throw new MaintenanceInputError("Invalid maintenance plan");
+  const value = input as Record<string, unknown>;
+  const reason = typeof value.reason === "string" ? value.reason.trim() : "";
+  const start = immediate
+    ? now
+    : typeof value.start === "string"
+      ? Date.parse(value.start)
+      : NaN;
+  const duration = value.durationMinutes;
+  const grace = value.graceMinutes;
+  const recurrence = immediate ? "once" : value.recurrence;
+  if (!reason || reason.length > 200)
+    throw new MaintenanceInputError("Reason must contain 1 to 200 characters");
+  if (
+    !Number.isFinite(start) ||
+    (!immediate && start <= now) ||
+    start > now + 366 * 10 * DAY
+  ) {
+    throw new MaintenanceInputError(
+      "Start must be a future time within ten years",
+    );
+  }
+  if (
+    typeof duration !== "number" ||
+    !Number.isInteger(duration) ||
+    duration < 1 ||
+    duration > 10080
+  ) {
+    throw new MaintenanceInputError(
+      "Estimated duration must be 1 to 10080 minutes",
+    );
+  }
+  if (
+    typeof grace !== "number" ||
+    !Number.isInteger(grace) ||
+    grace < 0 ||
+    grace > 1440
+  ) {
+    throw new MaintenanceInputError("Grace period must be 0 to 1440 minutes");
+  }
+  if (!["once", "weekly", "monthly"].includes(String(recurrence)))
+    throw new MaintenanceInputError("Invalid recurrence");
+  if (typeof value.notifyOverdue !== "boolean")
+    throw new MaintenanceInputError("Invalid overdue notification setting");
+  const iso = new Date(start).toISOString();
+  if (!immediate && start % 60_000 !== 0)
+    throw new MaintenanceInputError("Start must use whole minutes");
+  return {
+    id: "",
+    reason,
+    start: iso,
+    durationMinutes: duration,
+    graceMinutes: grace,
+    recurrence: recurrence as MaintenancePlan["recurrence"],
+    notifyOverdue: value.notifyOverdue,
+    nextStart: iso,
+  };
+}

plugins/automations/src/backend/scheduler.ts

@@ -23,6 +23,7 @@ export interface SchedulerDeps {
   repository: AutomationRepository;
   engine: Pick<AutomationEngine, "run">;
   log: PluginLogger;
+  isMaintaining?: (userId: string, hostId: number) => Promise<boolean>;
   /** Watchers brought in line with the enabled automations every tick. */
   reconcile: Array<(now: number) => Promise<unknown>>;
 }
@@ -78,6 +79,14 @@ export function createScheduler(deps: SchedulerDeps) {
     for (const state of open) {
       const automation = await repository.findById(state.automationId);
       if (!automation || !automation.enabled) continue;
+      const hostId = Number(state.stateKey.split(":")[0]);
+      if (
+        Number.isFinite(hostId) &&
+        (await deps.isMaintaining?.(automation.userId, hostId))
+      ) {
+        await repository.clearBreach(automation.id, state.stateKey);
+        continue;
+      }
 
       let definition: AutomationDefinition;
       try {
@@ -96,7 +105,6 @@ export function createScheduler(deps: SchedulerDeps) {
         continue;
       }
 
-      const hostId = Number(state.stateKey.split(":")[0]);
       await repository.upsertTriggerState({
         automationId: automation.id,
         stateKey: state.stateKey,

plugins/automations/src/backend/tables.ts

@@ -6,6 +6,7 @@ import {
   integer,
   real,
   refUser,
+  refHost,
   text,
   timestamp,
   varchar,
@@ -187,7 +188,23 @@ export const channels = adoptLegacyTable(
   ),
 );
 
+export const maintenanceTable = defineTable(
+  "host_maintenance",
+  {
+    id: id(),
+    userId: refUser(),
+    hostId: refHost(),
+    state: text().notNull(),
+  },
+  {
+    uniques: [
+      { name: "maintenance_host_owner", columns: ["userId", "hostId"] },
+    ],
+  },
+);
+
 export const tables = [
+  maintenanceTable,
   automations,
   triggerState,
   schedules,

plugins/automations/src/backend/triggers.ts

@@ -69,6 +69,10 @@ export function createTriggers(
   repository: AutomationRepository,
   engine: Pick<AutomationEngine, "run">,
   log: PluginLogger,
+  isMaintaining: (
+    userId: string,
+    hostId: number,
+  ) => Promise<boolean> = async () => false,
 ) {
   async function loadEnabledFor(userId: string): Promise<LoadedAutomation[]> {
     try {
@@ -120,6 +124,12 @@ export function createTriggers(
     triggerContext: Record<string, unknown>,
     hostId?: number,
   ): Promise<void> {
+    if (
+      hostId !== undefined &&
+      triggerType !== "internal_event" &&
+      (await isMaintaining(automation.userId, hostId))
+    )
+      return;
     await repository.upsertTriggerState({
       automationId: automation.id,
       stateKey,
@@ -149,6 +159,7 @@ export function createTriggers(
   async function onMetrics(event: MetricEvent): Promise<void> {
     const automations = await loadEnabledFor(event.ownerUserId);
     const now = Date.now();
+    const maintaining = await isMaintaining(event.ownerUserId, event.hostId);
 
     for (const { row, definition } of automations) {
       const trigger = definition.trigger;
@@ -162,7 +173,7 @@ export function createTriggers(
       const state = await repository.getTriggerState(row.id, stateKey);
       const breaching = compare(value, trigger.operator, trigger.value);
 
-      if (!breaching) {
+      if (maintaining || !breaching) {
         if (state?.breachStartedAt) {
           await repository.clearBreach(row.id, stateKey);
         }
@@ -228,6 +239,7 @@ export function createTriggers(
 
       // The first observation establishes a baseline rather than firing, so a
       // restart does not announce every host as though it just changed.
+      if (await isMaintaining(row.userId, event.hostId)) continue;
       if (!state?.lastObservedState) continue;
       if (trigger.to !== observed) continue;
       if (isCoolingDown(state?.lastFiredAt, trigger.cooldownMinutes, now))
@@ -265,6 +277,7 @@ export function createTriggers(
         lastObservedState: observed,
       });
 
+      if (await isMaintaining(row.userId, event.hostId)) continue;
       if (!state?.lastObservedState) continue;
       if (trigger.to !== observed) continue;
       if (isCoolingDown(state?.lastFiredAt, trigger.cooldownMinutes, now))
@@ -286,6 +299,7 @@ export function createTriggers(
   }
 
   async function onDockerEvent(event: DockerEvent): Promise<void> {
+    if (await isMaintaining(event.ownerUserId, event.hostId)) return;
     const automations = await loadEnabledFor(event.ownerUserId);
     const now = Date.now();
 
@@ -316,6 +330,12 @@ export function createTriggers(
   }
 
   async function onInternalEvent(event: InternalEvent): Promise<void> {
+    if (
+      event.event === "tunnel_disconnected" &&
+      event.hostId !== undefined &&
+      (await isMaintaining(event.userId, event.hostId))
+    )
+      return;
     const automations = await loadEnabledFor(event.userId);
     const now = Date.now();