/** * Instance-level plugin environment rows declared by the managed-config * `environments` section (`driver: "plugin"` entries, see `managed-config.ts`). * * Each declared entry owns exactly one row, identified by the marker * `metadata.managedPluginEnvironment = ":"` (unique * index `environments_managed_plugin_environment_idx`) next to * `managedByPaperclip: true`, which puts the row behind the environments-route * write floor on cloud-managed instances. * * - The first ensure adopts an existing unmarked row for the same plugin and * driver instead of adding a second one, so a hand-made row keeps the agents * that point at it. * - Later ensures converge the row to the declaration and write nothing when * it already matches. * - Archiving is only for a declared plugin that is not running; the next * successful ensure reactivates the row. Nothing here archives a row * because its entry left the document. * * The driver config always goes through the plugin environment validator * (`normalizeEnvironmentConfig` + `validatePluginEnvironmentDriverConfig`), * the same path an operator-created plugin environment takes. */ import { and, eq, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { companies, environments } from "@paperclipai/db"; import type { PluginEnvironmentConfig } from "@paperclipai/shared"; import { logActivity, publishActivity, type ActivityPublication } from "./activity-log.js"; import { normalizeEnvironmentConfig } from "./environment-config.js"; import { validatePluginEnvironmentDriverConfig } from "./plugin-environment-driver.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; export const MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY = "managedPluginEnvironment"; export function managedPluginEnvironmentMarker(input: { pluginKey: string; driverKey: string }): string { return `${input.pluginKey}:${input.driverKey}`; } export interface ManagedPluginEnvironmentInput { name: string; description?: string; pluginKey: string; driverKey: string; driverConfig?: Record; /** Runs the plugin's own config validation (`environmentValidateConfig`). */ workerManager: Pick; } export type ManagedPluginEnvironmentReconcileAction = "added" | "adopted" | "updated" | "unchanged"; export type EnvironmentRow = typeof environments.$inferSelect; export interface ManagedPluginEnvironmentRowResult { row: EnvironmentRow; action: ManagedPluginEnvironmentReconcileAction; } function metadataOf(row: Pick): Record { return (row.metadata ?? {}) as Record; } function stableJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`; if (value && typeof value === "object") { return `{${Object.keys(value as Record) .sort() .map((key) => `${JSON.stringify(key)}:${stableJson((value as Record)[key])}`) .join(",")}}`; } return JSON.stringify(value ?? null); } function servesPair(row: EnvironmentRow, input: { pluginKey: string; driverKey: string }): boolean { const config = (row.config ?? {}) as Record; return config.pluginKey === input.pluginKey && config.driverKey === input.driverKey; } async function validatedDriverConfig(db: Db, input: ManagedPluginEnvironmentInput): Promise { const normalized = normalizeEnvironmentConfig({ driver: "plugin", config: { pluginKey: input.pluginKey, driverKey: input.driverKey, driverConfig: input.driverConfig ?? {} }, }) as unknown as PluginEnvironmentConfig; return await validatePluginEnvironmentDriverConfig({ db, workerManager: input.workerManager as PluginWorkerManager, config: normalized, }); } /** Marker value of a row, or null when it has none (absent, or cleared to `null` / `false`). */ function markerOf(row: Pick): string | null { const value = metadataOf(row)[MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY]; return typeof value === "string" ? value : null; } /** * Record activity for every company inside the caller's transaction, so the * entries commit (or roll back) with the row change. Live events go to * `publications` and are published by the caller after commit. */ async function logForEveryCompany( tx: Db, publications: ActivityPublication[], input: { action: string; environmentId: string; details: Record }, ): Promise { const companyIds = await tx.select({ id: companies.id }).from(companies).then((rows) => rows.map((row) => row.id)); for (const companyId of companyIds) { await logActivity(tx, { companyId, actorType: "system", actorId: "managed-environment-reconciler", action: input.action, entityType: "environment", entityId: input.environmentId, details: input.details, }, publications); } } /** * Ensure the row for one declared plugin entry. Idempotent: a row that * already matches the declaration is returned without a write. */ export async function ensureManagedPluginEnvironment( db: Db, input: ManagedPluginEnvironmentInput, ): Promise { const marker = managedPluginEnvironmentMarker(input); const config = await validatedDriverConfig(db, input); const desiredConfig = { ...config } as Record; const description = input.description ?? null; const publications: ActivityPublication[] = []; const result = await db.transaction(async (tx): Promise => { const reconciled = await reconcile(tx); if (reconciled.action !== "unchanged") { await logForEveryCompany(tx as unknown as Db, publications, { action: `environment.managed_plugin_${reconciled.action}`, environmentId: reconciled.row.id, details: { pluginKey: input.pluginKey, driverKey: input.driverKey, reconciliationAction: reconciled.action }, }); } return reconciled; }); publications.forEach(publishActivity); return result; async function reconcile(tx: Parameters[0]>[0]): Promise { // Lock every plugin row: concurrent ensures (the boot pass and a worker // recovery, or two builds in a rolling deploy) serialize here, so only one // of them can adopt or insert. const pluginRows = await tx.select().from(environments).where(eq(environments.driver, "plugin")).for("update"); let row = pluginRows.find((candidate) => markerOf(candidate) === marker) ?? null; let adopted = false; if (!row) { const candidates = pluginRows // A cleared marker (`null` / `false`, from the marker-clear request) // counts as unmarked, so a cleared row is adopted again. .filter((candidate) => servesPair(candidate, input) && markerOf(candidate) === null) .sort((left, right) => Number(right.name === input.name) - Number(left.name === input.name) || Number(right.status === "active") - Number(left.status === "active") || left.createdAt.getTime() - right.createdAt.getTime()); row = candidates[0] ?? null; adopted = row !== null; } const nameOwner = await tx .select({ id: environments.id, driver: environments.driver }) .from(environments) .where(eq(environments.name, input.name)) .then((rows) => rows[0] ?? null); if (!row) { if (nameOwner) { throw new Error( `Failed to ensure managed plugin environment: environment "${input.name}" already exists with driver "${nameOwner.driver}"`, ); } const now = new Date(); const inserted = await tx .insert(environments) .values({ name: input.name, description, driver: "plugin", status: "active", config: desiredConfig, envVars: {}, metadata: { managedByPaperclip: true, [MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY]: marker }, createdAt: now, updatedAt: now, }) .onConflictDoNothing() .returning() .then((rows) => rows[0] ?? null); if (inserted) return { row: inserted, action: "added" }; // A concurrent pass won the marker index; converge on its row. row = await tx .select() .from(environments) .where(sql`${environments.metadata} ->> ${MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY} = ${marker}`) .for("update") .then((rows) => rows[0] ?? null); if (!row) throw new Error("Failed to ensure managed plugin environment"); } // Keep the row's own name when another row already holds the declared one. const desiredName = nameOwner && nameOwner.id !== row.id ? row.name : input.name; const currentMetadata = metadataOf(row); const desiredMetadata = { ...currentMetadata, managedByPaperclip: true, [MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY]: marker, }; const matches = row.name === desiredName && (row.description ?? null) === description && row.status === "active" && stableJson(row.config) === stableJson(desiredConfig) && stableJson(currentMetadata) === stableJson(desiredMetadata); if (matches) return { row, action: "unchanged" }; const updated = await tx .update(environments) .set({ name: desiredName, description, status: "active", config: desiredConfig, metadata: desiredMetadata, updatedAt: new Date(), }) .where(eq(environments.id, row.id)) .returning() .then((rows) => rows[0] ?? null); if (!updated) throw new Error("Managed plugin environment changed during reconciliation"); return { row: updated, action: adopted ? "adopted" : "updated" }; } } /** * Archive the marked row of a declared plugin that is not ready or has no * running worker, so run scheduling stops selecting it. Returns null when * there is no active marked row. */ export async function archiveManagedPluginEnvironment( db: Db, input: { pluginKey: string; driverKey: string }, ): Promise { const marker = managedPluginEnvironmentMarker(input); const publications: ActivityPublication[] = []; const archived = await db.transaction(async (tx) => { const row = await tx .update(environments) .set({ status: "archived", updatedAt: new Date() }) .where(and( eq(environments.driver, "plugin"), eq(environments.status, "active"), sql`${environments.metadata} ->> ${MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY} = ${marker}`, )) .returning() .then((rows) => rows[0] ?? null); if (!row) return null; await logForEveryCompany(tx as unknown as Db, publications, { action: "environment.managed_provider_unavailable_archived", environmentId: row.id, details: { pluginKey: input.pluginKey, driverKey: input.driverKey }, }); return row; }); publications.forEach(publishActivity); return archived; } /** The marked row for one plugin entry, archived or not. */ export async function findManagedPluginEnvironment( db: Db, input: { pluginKey: string; driverKey: string }, ): Promise { return await db .select() .from(environments) .where(sql`${environments.metadata} ->> ${MANAGED_PLUGIN_ENVIRONMENT_MARKER_KEY} = ${managedPluginEnvironmentMarker(input)}`) .then((rows) => rows[0] ?? null); }