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`);
};