feat: finish the docker plugin (B15)

plugins/automations/manifest.json

@@ -29,7 +29,8 @@
   "optionalDependencies": {
     "snippets": "^1.0.0",
     "fleets": "^1.0.0",
-    "tunnels": "^1.0.0"
+    "tunnels": "^1.0.0",
+    "docker": "^1.0.0"
   },
   "requires": [
     {
@@ -46,6 +47,16 @@
       "service": "tunnels.access",
       "versionRange": "^1.0.0",
       "optional": true
+    },
+    {
+      "service": "docker.containers",
+      "versionRange": "^1.0.0",
+      "optional": true
+    },
+    {
+      "service": "docker.events",
+      "versionRange": "^1.0.0",
+      "optional": true
     }
   ],
   "contributes": {

plugins/automations/src/backend/actions/index.ts

@@ -2,14 +2,11 @@ import {
   DEFAULT_STEP_TIMEOUT_MS,
   type Step,
 } from "../../../../../src/types/automations.js";
-import {
-  execCommand,
-  execElevated,
-  shellSingleQuote,
-} from "@termix/plugin-sdk/host-commands";
+import { execCommand, execElevated } from "@termix/plugin-sdk/host-commands";
 import { withSshConnection } from "../ssh.js";
 import { createCurrentNotificationChannelRepository } from "../../../../../src/backend/database/repositories/factory.js";
 import { getSnippet, resolveSnippetCommandFor } from "../snippets.js";
+import { runDockerAction } from "../docker.js";
 import { runTunnelAction } from "../tunnels.js";
 import { sendAutomationNotification } from "../notify.js";
 import { automationFetch } from "../http.js";
@@ -227,11 +224,15 @@ async function runDocker(
       };
     }
 
-    // The Docker HTTP routes are tied to an interactive session, so run the
-    // equivalent command over the same pooled SSH connection everything else
-    // uses. The container name is quoted because it comes from a template.
-    const command = `docker ${step.action} ${shellSingleQuote(container)}`;
-    return execOnHost(target.host, command, false, context, step.timeoutMs);
+    const result = await runDockerAction(
+      context.userId,
+      target.id,
+      container,
+      step.action,
+    );
+    return result.ok
+      ? { output: `Container ${container}: ${step.action} done` }
+      : { output: "", error: result.error };
   });
 }
 

plugins/automations/src/backend/docker-watcher.ts

@@ -1,41 +1,29 @@
 import type { AutomationDefinition } from "../../../../src/types/automations.js";
 import { createCurrentAutomationRepository } from "../../../../src/backend/database/repositories/factory.js";
-import { resolveHostById } from "../../../../src/backend/hosts/host-resolver.js";
-import { DataCrypto } from "../../../../src/backend/utils/data-crypto.js";
-import { execCommand } from "@termix/plugin-sdk/host-commands";
-import { withSshConnection } from "./ssh.js";
-import { statsLogger } from "../../../../src/backend/utils/logger.js";
+import { subscribeDockerEvents, type DockerEvent } from "./docker.js";
 import { onDockerEvent } from "./triggers.js";
 
 /**
- * Container state polling for docker_event triggers.
+ * Keeps a docker.events subscription for every host a docker_event trigger
+ * names, as the automation's owner. The docker plugin does the polling; this
+ * only follows which hosts are watched.
  *
- * Everything else Docker-related in the backend hangs off an interactive
- * session that only exists while somebody has the UI open, so a trigger built
- * on it would only ever fire while being watched. This polls over the same
- * pooled SSH connection the other automation steps use, and only for hosts a
- * docker_event trigger actually names, so an install with no such automation
- * does no extra work at all.
- *
- * Events are derived by diffing successive snapshots: the poll interval is the
- * resolution, so a container that stops and starts between two polls is not
- * reported. That is the tradeoff for not holding a `docker events` stream open
- * against every host.
+ * Subscribing is idempotent on the docker side, so every watched host is
+ * subscribed again once a minute. That is what picks the subscriptions back
+ * up after the docker plugin is disabled and enabled again.
  */
 
-const POLL_INTERVAL_MS = 60_000;
-const EXEC_TIMEOUT_MS = 15_000;
+const RESUBSCRIBE_MS = 60_000;
 
-interface ContainerState {
-  /** Docker's own state word: running, exited, restarting, ... */
-  state: string;
-  /** Health from the status text, when the image declares a healthcheck. */
-  unhealthy: boolean;
+interface Watch {
+  listener: (event: DockerEvent) => void;
+  unsubscribe: (() => void) | null;
+  subscribedAt: number;
 }
 
-/** Last snapshot per host, so transitions can be spotted. */
-const snapshots = new Map<number, Map<string, ContainerState>>();
-const lastPolledAt = new Map<number, number>();
+const watches = new Map<string, Watch>();
+
+const keyOf = (hostId: number, userId: string) => `${userId}:${hostId}`;
 
 /** Hosts named by an enabled docker_event trigger, with the owning user. */
 export async function listDockerWatchedHosts(): Promise<Map<number, string>> {
@@ -60,8 +48,8 @@ export async function listDockerWatchedHosts(): Promise<Map<number, string>> {
       } else if (selector?.kind === "hosts") {
         for (const hostId of selector.hostIds) watched.set(hostId, row.userId);
       }
-      // Fleet and "all" selectors are deliberately not expanded: polling every
-      // host a user owns for container state is far too costly to do blindly.
+      // Fleet and "all" selectors are deliberately not expanded: watching
+      // every host a user owns for container state is far too costly.
     }
   } catch {
     return watched;
@@ -70,165 +58,58 @@ export async function listDockerWatchedHosts(): Promise<Map<number, string>> {
   return watched;
 }
 
-/**
- * Parses `docker ps -a` output. One JSON object per line, matching the format
- * string the container routes use.
- */
-export function parseContainerStates(
-  output: string,
-): Map<string, ContainerState> {
-  const states = new Map<string, ContainerState>();
-
-  for (const line of output.split("\n")) {
-    const trimmed = line.trim();
-    if (!trimmed) continue;
-
-    try {
-      const parsed = JSON.parse(trimmed) as {
-        name?: string;
-        state?: string;
-        status?: string;
-      };
-      if (!parsed.name) continue;
-
-      states.set(parsed.name, {
-        state: (parsed.state ?? "").toLowerCase(),
-        unhealthy: /\(unhealthy\)/i.test(parsed.status ?? ""),
-      });
-    } catch {
-      // A partial line is not worth failing the whole poll over.
-    }
-  }
-
-  return states;
-}
-
-/**
- * Works out which events a pair of snapshots implies.
- *
- * A container missing from the previous snapshot is treated as newly seen
- * rather than started, so the first poll after a restart does not replay every
- * running container as a fresh start event.
- */
-export function diffContainerStates(
-  previous: Map<string, ContainerState>,
-  current: Map<string, ContainerState>,
-): Array<{ container: string; event: DockerEventName }> {
-  const events: Array<{ container: string; event: DockerEventName }> = [];
-
-  for (const [name, now] of current) {
-    const before = previous.get(name);
-    if (!before) continue;
-
-    if (before.state !== now.state) {
-      if (now.state === "exited")
-        events.push({ container: name, event: "exited" });
-      else if (now.state === "running")
-        events.push({ container: name, event: "started" });
-      else if (now.state === "restarting")
-        events.push({ container: name, event: "restarting" });
-    }
-
-    // Health is independent of state: a container can go unhealthy while it
-    // stays up, which is exactly the case worth alerting on.
-    if (!before.unhealthy && now.unhealthy) {
-      events.push({ container: name, event: "unhealthy" });
-    }
-  }
-
-  return events;
-}
-
-export type DockerEventName = "exited" | "started" | "unhealthy" | "restarting";
-
-const PS_FORMAT = `'{"name":"{{.Names}}","state":"{{.State}}","status":"{{.Status}}"}'`;
-
-async function pollHost(hostId: number, userId: string): Promise<void> {
-  const host = await resolveHostById(hostId, userId);
-  if (!host) {
-    snapshots.delete(hostId);
-    return;
-  }
-
-  const result = await withSshConnection(
-    host as never as never,
-    { pool: "fleet", purpose: "fleet" },
-    (client) =>
-      execCommand(
-        client,
-        `docker ps -a --format ${PS_FORMAT}`,
-        EXEC_TIMEOUT_MS,
-      ),
+/** Brings the subscriptions in line with the watched hosts. */
+export async function reconcileDockerWatch(
+  now: number = Date.now(),
+  watched?: Map<number, string>,
+): Promise<void> {
+  const wanted = watched ?? (await listDockerWatchedHosts());
+  const wantedKeys = new Set(
+    [...wanted].map(([hostId, userId]) => keyOf(hostId, userId)),
   );
 
-  if (result.code !== 0 && result.code !== null) {
-    // Docker missing or not permitted on this host. Drop the snapshot so a
-    // later success is treated as a first observation rather than a diff.
-    snapshots.delete(hostId);
-    return;
+  for (const [key, watch] of watches) {
+    if (wantedKeys.has(key)) continue;
+    watch.unsubscribe?.();
+    watches.delete(key);
   }
 
-  const current = parseContainerStates(result.stdout);
-  const previous = snapshots.get(hostId);
-  snapshots.set(hostId, current);
-
-  if (!previous) return;
-
-  for (const { container, event } of diffContainerStates(previous, current)) {
-    await onDockerEvent({
-      hostId,
-      ownerUserId: userId,
-      container,
-      event,
-    }).catch(() => undefined);
-  }
-}
-
-/**
- * Polls every watched host whose interval has elapsed. Called from the
- * automation scheduler tick rather than owning a timer of its own.
- */
-export async function pollDockerEvents(
-  now: number = Date.now(),
-): Promise<void> {
-  const watched = await listDockerWatchedHosts();
-
-  for (const hostId of [...snapshots.keys()]) {
-    if (!watched.has(hostId)) {
-      snapshots.delete(hostId);
-      lastPolledAt.delete(hostId);
+  for (const [hostId, userId] of wanted) {
+    const key = keyOf(hostId, userId);
+    let watch = watches.get(key);
+    if (!watch) {
+      watch = {
+        listener: (event) => {
+          void onDockerEvent({
+            hostId: event.hostId,
+            ownerUserId: userId,
+            container: event.container,
+            event: event.event,
+          }).catch(() => undefined);
+        },
+        unsubscribe: null,
+        subscribedAt: 0,
+      };
+      watches.set(key, watch);
+    }
+    if (watch.subscribedAt && now - watch.subscribedAt < RESUBSCRIBE_MS) {
+      continue;
     }
-  }
-
-  for (const [hostId, userId] of watched) {
-    const last = lastPolledAt.get(hostId) ?? 0;
-    if (now - last < POLL_INTERVAL_MS) continue;
-    // Host credentials cannot be decrypted while the owner's key is locked.
-    if (!canAccess(userId)) continue;
-    lastPolledAt.set(hostId, now);
-
     try {
-      await pollHost(hostId, userId);
-    } catch (error) {
-      statsLogger.warn("Docker event poll failed", {
-        operation: "automation_docker_poll_error",
+      watch.unsubscribe = await subscribeDockerEvents(
+        userId,
         hostId,
-        error: error instanceof Error ? error.message : String(error),
-      });
+        watch.listener,
+      );
+      watch.subscribedAt = watch.unsubscribe ? now : 0;
+    } catch {
+      watch.subscribedAt = 0;
     }
   }
 }
 
-function canAccess(userId: string): boolean {
-  try {
-    return DataCrypto.canUserAccessData(userId);
-  } catch {
-    return false;
-  }
-}
-
-/** Clears cached state, for shutdown and tests. */
+/** Drops every subscription, for shutdown and tests. */
 export function resetDockerWatcher(): void {
-  snapshots.clear();
-  lastPolledAt.clear();
+  for (const watch of watches.values()) watch.unsubscribe?.();
+  watches.clear();
 }

plugins/automations/src/backend/docker.ts

@@ -0,0 +1,85 @@
+import type { PluginServices } from "@termix/plugin-sdk/backend";
+
+let current: PluginServices | null = null;
+
+/** Set in activate, cleared on deactivate. */
+export function setDockerServices(services: PluginServices | null): void {
+  current = services;
+}
+
+export type DockerEventName = "exited" | "started" | "unhealthy" | "restarting";
+
+export interface DockerEvent {
+  hostId: number;
+  container: string;
+  event: DockerEventName;
+}
+
+/** The docker plugin's "docker.containers" service, version 1. */
+interface DockerService {
+  action: (
+    hostId: number,
+    container: string,
+    action: "start" | "stop" | "restart",
+  ) => Promise<void>;
+}
+
+/** The docker plugin's "docker.events" service, version 1. */
+interface DockerEventsService {
+  subscribe: (
+    hostId: number,
+    listener: (event: DockerEvent) => void,
+  ) => Promise<() => void>;
+}
+
+const UNAVAILABLE = "The docker plugin is not available";
+
+function service<T extends object>(
+  name: string,
+  userId: string,
+): Partial<T> | null {
+  try {
+    return current?.get<T>(name, { userId }) ?? null;
+  } catch {
+    return null;
+  }
+}
+
+/**
+ * Starts, stops or restarts a container as userId. Optional: without the
+ * docker plugin the step fails with a clear message.
+ */
+export async function runDockerAction(
+  userId: string,
+  hostId: number,
+  container: string,
+  action: "start" | "stop" | "restart",
+): Promise<{ ok: boolean; unavailable?: boolean; error?: string }> {
+  const docker = service<DockerService>("docker.containers", userId);
+  if (!docker || typeof docker.action !== "function") {
+    return { ok: false, unavailable: true, error: UNAVAILABLE };
+  }
+  try {
+    await docker.action(hostId, container, action);
+    return { ok: true };
+  } catch (error) {
+    return {
+      ok: false,
+      error: error instanceof Error ? error.message : String(error),
+    };
+  }
+}
+
+/**
+ * Subscribes `listener` to container changes on a host as userId, or
+ * resolves null while the docker plugin is off.
+ */
+export async function subscribeDockerEvents(
+  userId: string,
+  hostId: number,
+  listener: (event: DockerEvent) => void,
+): Promise<(() => void) | null> {
+  const events = service<DockerEventsService>("docker.events", userId);
+  if (!events || typeof events.subscribe !== "function") return null;
+  return events.subscribe(hostId, listener);
+}

plugins/automations/src/backend/index.ts

@@ -2,6 +2,7 @@ import type { PluginContext } from "@termix/plugin-sdk/backend";
 import { setPluginSsh } from "./ssh.js";
 import { setPluginServices } from "./snippets.js";
 import { setTunnelServices } from "./tunnels.js";
+import { setDockerServices } from "./docker.js";
 import { startAutomationsService, stopAutomationsService } from "./routes.js";
 import {
   startAutomationScheduler,
@@ -23,6 +24,8 @@ export async function activate(ctx: PluginContext) {
   ctx.disposables.add(() => setPluginServices(null));
   setTunnelServices(ctx.services);
   ctx.disposables.add(() => setTunnelServices(null));
+  setDockerServices(ctx.services);
+  ctx.disposables.add(() => setDockerServices(null));
 
   // Nothing arrives here while the tunnels plugin is off, which is the whole
   // of the optional dependency.

plugins/automations/src/backend/scheduler.ts

@@ -4,7 +4,7 @@ import { DataCrypto } from "../../../../src/backend/utils/data-crypto.js";
 import { statsLogger } from "../../../../src/backend/utils/logger.js";
 import { computeNextDueAt } from "./cron.js";
 import { hasDwelled, isCoolingDown } from "./conditions.js";
-import { pollDockerEvents } from "./docker-watcher.js";
+import { reconcileDockerWatch, resetDockerWatcher } from "./docker-watcher.js";
 import { AutomationEngine } from "./engine.js";
 import { reconcileHeadlessViewers } from "./headless-viewer.js";
 import {
@@ -57,6 +57,7 @@ export function stopAutomationScheduler(): void {
   tickTimer = null;
   startupTimer = null;
   unsubscribeAutomationTriggers();
+  resetDockerWatcher();
 }
 
 /** Exposed for tests; the interval calls this. */
@@ -69,7 +70,7 @@ export async function tick(now: Date = new Date()): Promise<void> {
     await reconcileHeadlessViewers().catch(() => undefined);
     await runDueSchedules(now);
     await recheckOpenBreaches(now);
-    await pollDockerEvents(now.getTime()).catch(() => undefined);
+    await reconcileDockerWatch(now.getTime()).catch(() => undefined);
     await pruneIfDue(now);
   } catch (error) {
     statsLogger.warn("Automation scheduler tick failed", {

plugins/automations/tests/backend/docker-watcher.test.ts

@@ -1,108 +1,131 @@
-import { describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { PluginServices } from "@termix/plugin-sdk/backend";
+
+const fired = vi.hoisted(() => [] as unknown[]);
+
+vi.mock("../../src/backend/triggers.js", () => ({
+  onDockerEvent: async (event: unknown) => {
+    fired.push(event);
+  },
+}));
+
+vi.mock("../../../../src/backend/database/repositories/factory.js", () => ({
+  createCurrentAutomationRepository: () => ({ listAllEnabled: async () => [] }),
+}));
+
 import {
-  diffContainerStates,
-  parseContainerStates,
+  reconcileDockerWatch,
+  resetDockerWatcher,
 } from "../../src/backend/docker-watcher.js";
+import {
+  setDockerServices,
+  runDockerAction,
+} from "../../src/backend/docker.js";
 
-/**
- * The polling side needs SSH, so what is tested here is the pure part: turning
- * `docker ps` output into states, and turning two snapshots into events.
- */
+type Listener = (event: {
+  hostId: number;
+  container: string;
+  event: string;
+}) => void;
 
-describe("parseContainerStates", () => {
-  it("reads name, state and health out of the ps output", () => {
-    const output = [
-      '{"name":"web","state":"running","status":"Up 2 hours"}',
-      '{"name":"db","state":"exited","status":"Exited (0) 5 minutes ago"}',
-      '{"name":"api","state":"running","status":"Up 1 hour (unhealthy)"}',
-    ].join("\n");
+/** A docker.events provider that records who subscribed to what. */
+function fakeEvents() {
+  const subscriptions = new Map<string, Set<Listener>>();
+  const calls: Array<{ userId?: string; hostId: number }> = [];
+  const services = {
+    provide: () => {},
+    providers: () => [],
+    get: (name: string, options?: { userId?: string }) => {
+      if (name !== "docker.events") return {};
+      return {
+        subscribe: async (hostId: number, listener: Listener) => {
+          calls.push({ userId: options?.userId, hostId });
+          const key = `${options?.userId}:${hostId}`;
+          const set = subscriptions.get(key) ?? new Set<Listener>();
+          set.add(listener);
+          subscriptions.set(key, set);
+          return () => set.delete(listener);
+        },
+      };
+    },
+  } as unknown as PluginServices;
+  const emit = (key: string, event: Parameters<Listener>[0]) => {
+    for (const listener of subscriptions.get(key) ?? []) listener(event);
+  };
+  return { services, subscriptions, calls, emit };
+}
 
-    const states = parseContainerStates(output);
-
-    expect(states.get("web")).toEqual({ state: "running", unhealthy: false });
-    expect(states.get("db")).toEqual({ state: "exited", unhealthy: false });
-    expect(states.get("api")).toEqual({ state: "running", unhealthy: true });
-  });
-
-  it("skips blank and malformed lines rather than failing the poll", () => {
-    const output = [
-      '{"name":"web","state":"running","status":"Up"}',
-      "",
-      "not json at all",
-      '{"state":"running"}',
-    ].join("\n");
-
-    const states = parseContainerStates(output);
-    expect([...states.keys()]).toEqual(["web"]);
-  });
-
-  it("returns nothing for empty output", () => {
-    expect(parseContainerStates("").size).toBe(0);
-  });
+beforeEach(() => {
+  fired.length = 0;
 });
 
-describe("diffContainerStates", () => {
-  const running = { state: "running", unhealthy: false };
-  const exited = { state: "exited", unhealthy: false };
+afterEach(() => {
+  resetDockerWatcher();
+  setDockerServices(null);
+});
 
-  it("reports a container that stopped", () => {
-    const events = diffContainerStates(
-      new Map([["web", running]]),
-      new Map([["web", exited]]),
-    );
-    expect(events).toEqual([{ container: "web", event: "exited" }]);
+describe("docker watcher", () => {
+  it("subscribes each watched host as its owner and fires the trigger", async () => {
+    const events = fakeEvents();
+    setDockerServices(events.services);
+
+    await reconcileDockerWatch(0, new Map([[7, "user-1"]]));
+    expect(events.calls).toEqual([{ userId: "user-1", hostId: 7 }]);
+
+    events.emit("user-1:7", { hostId: 7, container: "web", event: "exited" });
+    await Promise.resolve();
+    expect(fired).toEqual([
+      { hostId: 7, ownerUserId: "user-1", container: "web", event: "exited" },
+    ]);
   });
 
-  it("reports a container that started", () => {
-    const events = diffContainerStates(
-      new Map([["web", exited]]),
-      new Map([["web", running]]),
-    );
-    expect(events).toEqual([{ container: "web", event: "started" }]);
+  it("re-subscribes once a minute and drops hosts no longer watched", async () => {
+    const events = fakeEvents();
+    setDockerServices(events.services);
+    const watched = new Map([[7, "user-1"]]);
+
+    await reconcileDockerWatch(1_000, watched);
+    await reconcileDockerWatch(30_000, watched);
+    expect(events.calls).toHaveLength(1);
+    await reconcileDockerWatch(62_000, watched);
+    expect(events.calls).toHaveLength(2);
+    expect(events.subscriptions.get("user-1:7")?.size).toBe(1);
+
+    await reconcileDockerWatch(63_000, new Map());
+    expect(events.subscriptions.get("user-1:7")?.size).toBe(0);
   });
 
-  it("reports a container that began restarting", () => {
-    const events = diffContainerStates(
-      new Map([["web", running]]),
-      new Map([["web", { state: "restarting", unhealthy: false }]]),
-    );
-    expect(events).toEqual([{ container: "web", event: "restarting" }]);
+  it("does nothing while the docker plugin is off, then picks it up", async () => {
+    setDockerServices({
+      get: () => ({}),
+    } as unknown as PluginServices);
+    await expect(
+      reconcileDockerWatch(0, new Map([[7, "user-1"]])),
+    ).resolves.toBeUndefined();
+
+    const events = fakeEvents();
+    setDockerServices(events.services);
+    await reconcileDockerWatch(1, new Map([[7, "user-1"]]));
+    expect(events.calls).toHaveLength(1);
   });
 
-  it("reports a container that went unhealthy while still running", () => {
-    const events = diffContainerStates(
-      new Map([["web", running]]),
-      new Map([["web", { state: "running", unhealthy: true }]]),
-    );
-    expect(events).toEqual([{ container: "web", event: "unhealthy" }]);
+  it("fails a docker step cleanly while the plugin is off", async () => {
+    setDockerServices({ get: () => ({}) } as unknown as PluginServices);
+    expect(await runDockerAction("user-1", 7, "web", "restart")).toEqual({
+      ok: false,
+      unavailable: true,
+      error: "The docker plugin is not available",
+    });
   });
 
-  it("stays quiet when nothing changed", () => {
-    const events = diffContainerStates(
-      new Map([["web", running]]),
-      new Map([["web", running]]),
-    );
-    expect(events).toEqual([]);
-  });
-
-  it("does not re-announce a container that is still unhealthy", () => {
-    const unhealthy = { state: "running", unhealthy: true };
-    const events = diffContainerStates(
-      new Map([["web", unhealthy]]),
-      new Map([["web", unhealthy]]),
-    );
-    expect(events).toEqual([]);
-  });
-
-  // A first sighting is a baseline, not an event: otherwise every container
-  // running at boot would report itself as freshly started.
-  it("treats a newly seen container as a baseline", () => {
-    const events = diffContainerStates(new Map(), new Map([["web", running]]));
-    expect(events).toEqual([]);
-  });
-
-  it("ignores a container that disappeared", () => {
-    const events = diffContainerStates(new Map([["web", running]]), new Map());
-    expect(events).toEqual([]);
+  it("runs a docker step through the docker service", async () => {
+    const action = vi.fn(async () => {});
+    setDockerServices({
+      get: (name: string) => (name === "docker.containers" ? { action } : {}),
+    } as unknown as PluginServices);
+    expect(await runDockerAction("user-1", 7, "web", "restart")).toEqual({
+      ok: true,
+    });
+    expect(action).toHaveBeenCalledWith(7, "web", "restart");
   });
 });