camino-package-runtime/src/migration.ts

import { createHash } from "node:crypto";

export type MigrationInput = {
  schemaVersion: 1;
  executionId: string;
  exportId: string;
  ports: {
    name: string;
    binding: string;
    view: "old" | "new";
    access: ("read" | "write" | "create" | "edge")[];
    atomId?: string;
    attachedAtomId?: string;
    defaultValue?: unknown;
    states?: { objectId: string; value: unknown }[];
    edges?: MigrationEdge[];
  }[];
};
export type MigrationEdge = {
  id: string;
  edgeTypeId: string;
  firstObjectId: string;
  secondObjectId: string;
  firstProjectionId: string;
  secondProjectionId: string;
  firstOrdinal?: number;
  secondOrdinal?: number;
  firstKeyJson?: string;
  secondKeyJson?: string;
};
export type MigrationOutput = {
  schemaVersion: 1;
  executionId: string;
  writes: { port: string; objectId: string; value: unknown }[];
  creates: { port: string; logicalKey: string; objectId: string }[];
  edgeReplacements: { port: string; edges: MigrationEdge[] }[];
};
export type MigrationContext = {
  enumerate(port: string): { objectId: string; value: unknown }[];
  read(port: string, objectId: string): unknown;
  write(port: string, objectId: string, value: unknown): void;
  create(port: string, logicalKey: string): string;
  edges(port: string): MigrationEdge[];
  replaceEdges(port: string, edges: MigrationEdge[]): void;
};
export const migrationObjectId = (executionId: string, port: string, logicalKey: string) =>
  `obj:migration:${createHash("sha256")
    .update(JSON.stringify([executionId, port, logicalKey]))
    .digest("hex")}`;

/** No ordinary RuntimeContext or network/database clients are supplied here.
 * Process isolation belongs to the host, not this convenience API. */
export const createMigrationContext = (input: MigrationInput) => {
  if (
    input.schemaVersion !== 1 ||
    !input.executionId ||
    new Set(input.ports.map((entry) => entry.name)).size !== input.ports.length
  )
    throw new Error("Invalid migration input");
  const output: MigrationOutput = {
    schemaVersion: 1,
    executionId: input.executionId,
    writes: [],
    creates: [],
    edgeReplacements: [],
  };
  const port = (name: string, access: string) => {
    const selected = input.ports.find((entry) => entry.name === name);
    if (!selected?.access.includes(access as "read") || (selected.view === "old" && access !== "read"))
      throw new Error(`Migration port ${name} does not grant ${access}`);
    return selected;
  };
  const context: MigrationContext = {
    enumerate(name) {
      const selected = port(name, "read"),
        states = structuredClone(selected.states ?? []);
      if (selected.view === "new" && Object.hasOwn(selected, "defaultValue"))
        for (const helper of output.creates) {
          if (
            input.ports.find((entry) => entry.name === helper.port)?.atomId === selected.attachedAtomId &&
            !states.some((entry) => entry.objectId === helper.objectId)
          )
            states.push({ objectId: helper.objectId, value: structuredClone(selected.defaultValue) });
        }
      if (selected.view === "new")
        for (const write of output.writes) {
          if (input.ports.find((entry) => entry.name === write.port)?.binding !== selected.binding) continue;
          const existing = states.findIndex((entry) => entry.objectId === write.objectId),
            entry = { objectId: write.objectId, value: structuredClone(write.value) };
          if (existing < 0) states.push(entry);
          else states[existing] = entry;
        }
      return states.sort((a, b) => (a.objectId < b.objectId ? -1 : a.objectId > b.objectId ? 1 : 0));
    },
    read(name, objectId) {
      return context.enumerate(name).find((entry) => entry.objectId === objectId)?.value;
    },
    write(name, objectId, value) {
      port(name, "write");
      const previous = output.writes.findIndex((entry) => entry.port === name && entry.objectId === objectId);
      const entry = { port: name, objectId, value: structuredClone(value) };
      if (previous < 0) output.writes.push(entry);
      else output.writes[previous] = entry;
    },
    create(name, logicalKey) {
      port(name, "create");
      if (!logicalKey || logicalKey.length > 1024)
        throw new Error("Migration creation requires a bounded stable logical key");
      const objectId = migrationObjectId(input.executionId, name, logicalKey);
      if (!output.creates.some((entry) => entry.objectId === objectId))
        output.creates.push({ port: name, logicalKey, objectId });
      return objectId;
    },
    edges(name) {
      const selected = port(name, "read");
      const replacement =
        selected.view === "new"
          ? output.edgeReplacements.find(
              (entry) => input.ports.find((candidate) => candidate.name === entry.port)?.binding === selected.binding,
            )
          : undefined;
      return structuredClone(replacement?.edges ?? selected.edges ?? []);
    },
    replaceEdges(name, edges) {
      port(name, "edge");
      const previous = output.edgeReplacements.findIndex((entry) => entry.port === name);
      const entry = { port: name, edges: structuredClone(edges) };
      if (previous < 0) output.edgeReplacements.push(entry);
      else output.edgeReplacements[previous] = entry;
    },
  };
  return { context, result: () => structuredClone(output) };
};

/** Entrypoint for an immutable package's dedicated bin/migrate executable.
 * stdout is protocol-only; send diagnostics to stderr. The host independently
 * validates every write, helper identity, contract, and completion receipt. */
export const serveMigration = async (exports: Record<string, (context: MigrationContext) => void | Promise<void>>) => {
  const chunks: Buffer[] = [];
  let bytes = 0;
  for await (const chunk of process.stdin) {
    bytes += chunk.length;
    if (bytes > 16 * 1024 * 1024) throw new Error("Migration input exceeds 16 MiB");
    chunks.push(Buffer.from(chunk));
  }
  const input = JSON.parse(Buffer.concat(chunks).toString("utf8")) as MigrationInput;
  const implementation = Object.hasOwn(exports, input.exportId) ? exports[input.exportId] : undefined;
  if (!implementation) throw new Error("Unknown migration export");
  const execution = createMigrationContext(input);
  await implementation(execution.context);
  const result = JSON.stringify(execution.result());
  if (Buffer.byteLength(result) > 16 * 1024 * 1024) throw new Error("Migration output exceeds 16 MiB");
  process.stdout.write(`${result}\n`);
};