quixos-protocol/test/journals.test.ts

import assert from "node:assert/strict";
import { test } from "node:test";
import { compileCapabilityResourceSource, compileCapabilitySource } from "../src/capability-language/parser.js";
import { formatQx } from "../src/capability-language/source.js";
import { compileExecutionPlan, compileKernelProgram } from "../src/execution-world/index.js";
import type { InterfaceRevision } from "../src/capability-model/types.js";
import { readFileSync } from "node:fs";

const source = { repository: "https://example.test/journals", commit: "a".repeat(40) };

test("standard event contracts instantiate generic native conformance rules", () => {
  const interfaces = new Map<string, InterfaceRevision>();
  for (const [name, slug] of [
    ["EventStream", "event-stream"],
    ["EventHandler", "event-handler"],
    ["EventSubscription", "event-subscription"],
    ["EventStreamRecord", "event-stream-record"],
    ["EventSubscriptionRecord", "event-subscription-record"],
  ]) {
    const text = readFileSync(new URL(`../../test/fixtures/events/${slug}.qx`, import.meta.url), "utf8");
    const result = compileCapabilityResourceSource(text, {
      source,
      environment: { interfaces, interfaceClosure: [...interfaces.values()] },
    });
    assert.ok(result.ok && result.resource.kind === "interface", JSON.stringify(result.diagnostics));
    interfaces.set(name!, result.resource.revision);
  }
  const rules = readFileSync(new URL("../../test/fixtures/events/rules.qx", import.meta.url), "utf8")
    .replace(/^fragment\s*\{/, "")
    .replace(/}\s*$/, "");
  const result = compileCapabilitySource(
    `workspace Events id "events" revision "events@1" commit "${source.commit}" {
    ${rules}
    atom TextStream id "text-stream";
    conform TextStream as EventStreamRecord<string> id "text-marker" {}
    atom TextSubscription id "text-subscription";
    conform TextSubscription as EventSubscriptionRecord<string,unit> id "text-subscription-marker" {}
    atom NumberStream id "number-stream";
    conform NumberStream as EventStreamRecord<uint32> id "number-marker" {}
  }`,
    "events.qx",
    { interfaces, interfaceClosure: [...interfaces.values()] },
  );
  assert.ok(result.ok, JSON.stringify(result.diagnostics));
  const kernel = compileKernelProgram(compileExecutionPlan(result.workspace)) as {
    model: { owners: { slots: { policy: string }[] }[] };
  };
  assert.equal(
    kernel.model.owners.flatMap((owner) => owner.slots).filter((slot) => slot.policy === "journal").length,
    2,
  );
  assert.equal(
    kernel.model.owners.flatMap((owner) => owner.slots).filter((slot) => slot.policy === "subscription").length,
    1,
  );
});
const contract = `external atom Document id "document";
type Event = record { document: atom-ref<Document>; title: string; };
interface Stream id "stream" revision "stream@1" {
  operation publish id "publish" : Event -> record { publication: string; } {
    call id "publish:call" execution transaction;
  }
  queryable value head id "head" : optional<uint64> { get id "head:get" execution snapshot; }
  operation readPage id "page" : record { after: optional<string>; limit: uint32; }
    -> record { entries: list<record { publication: string; position: uint64; payload: Event; }>; next: optional<string>; asOfHead: optional<uint64>; } {
    call id "page:call" execution snapshot;
  }
}`;
const assembly = `workspace Journals id "journals" revision "journals@1" commit "${source.commit}" {
  import interface Stream;
  atom Document id "document";
  atom Journal id "journal-object";
  conform Journal as Stream id "journal-owner" {
    private state Entries id "entries" on Journal : record { document: atom-ref<Document>; title: string; }
      policy journal { "maxEntries": 32, "maxBytes": 65536, "maxPublicationBytes": 8192, "maxProtectedSubscriptions": 8 };
    bind publish.call to journal Entries.publish;
    bind head.get to journal Entries.head;
    bind readPage.call to journal Entries.readPage;
  }
}`;
function compile(text = assembly, iface = contract) {
  const resource = compileCapabilityResourceSource(iface, { source });
  assert.ok(resource.ok && resource.resource.kind === "interface", JSON.stringify(resource.diagnostics));
  return compileCapabilitySource(formatQx(text), "journal.qx", {
    interfaces: new Map([["Stream", resource.resource.revision]]),
  });
}
test("journal bindings typecheck and lower managed reference pins", () => {
  const result = compile();
  assert.ok(result.ok, JSON.stringify(result.diagnostics));
  const kernel = compileKernelProgram(compileExecutionPlan(result.workspace)) as {
    model: { edges: unknown[]; owners: { slots: { policy: string; journal: { pins: unknown[] } }[] }[] };
    operations: { binding: { kind: string; primitive: string } }[];
  };
  assert.equal(kernel.model.owners[0]!.slots[0]!.policy, "journal");
  assert.equal(kernel.model.owners[0]!.slots[0]!.journal.pins.length, 1);
  assert.equal(kernel.model.edges.length, 1);
  assert.deepEqual(kernel.operations.map((op) => op.binding.primitive).sort(), ["head", "publish", "readPage"]);
  assert.ok(kernel.operations.every((op) => op.binding.kind === "journal"));
});
test("journal declarations reject state access, bad bounds, seeding and signature substitution", () => {
  for (const [ws, iface] of [
    [assembly.replace("journal Entries.head", "state Entries.read"), contract],
    [assembly.replace('"maxEntries": 32', '"maxEntries": 0'), contract],
    [assembly.replace("bind publish.call", "initialize { seed to state Entries; }; bind publish.call"), contract],
    [assembly, contract.replace("publication: string", "publication: uint32")],
    [assembly, contract.replace("execution transaction", "execution snapshot")],
  ]) {
    const result = compile(ws, iface);
    assert.equal(result.ok, false, `${ws}\n${iface}`);
  }
});

test("subscription storage infers its typed handler and retains configuration references", () => {
  const interfaces = new Map<string, InterfaceRevision>();
  const resource = (name: string, text: string) => {
    const result = compileCapabilityResourceSource(text, {
      source,
      environment: { interfaces, interfaceClosure: [...interfaces.values()] },
    });
    assert.ok(result.ok && result.resource.kind === "interface", JSON.stringify(result.diagnostics));
    interfaces.set(name, result.resource.revision);
  };
  resource("Stream", contract);
  resource(
    "Receiver",
    `external atom Document id "document";
interface Receiver id "receiver" revision "receiver@1" {
  operation receive id "receive" : record { event: record { document: atom-ref<Document>; title: string; }; parameters: string; } -> unit {
    call id "receive:call" execution transaction;
  }
}`,
  );
  resource(
    "Subscription",
    `external atom Document id "document";
import interface Stream;
import interface Receiver;
type Start = variant id "start" { future id "future" : unit; retained id "retained" : unit; afterCursor id "after" : string; };
interface Subscription id "subscription" revision "subscription@1" {
  operation attach id "attach" : record { source: interface-ref<Stream>; handler: interface-ref<Receiver>; parameters: string; start: Start; } -> unit {
    call id "attach:call" execution transaction;
  }
  queryable value status id "status" : record { status: string; generation: uint64; cursor: uint64; failure: optional<record { publication: string; code: string; message: string; }>; } {
    get id "status:get" execution snapshot;
  }
  operation retire id "retire" : unit -> unit { call id "retire:call" execution transaction; }
}`,
  );
  const ws = assembly
    .replace(
      "import interface Stream;",
      "import interface Stream; import interface Receiver; import interface Subscription;",
    )
    .replace(
      /\n}$/,
      `
  atom Delivery id "delivery";
  conform Delivery as Subscription id "delivery-owner" {
    private state DeliveryState id "delivery-state" on Delivery : string policy subscription(record { document: atom-ref<Document>; title: string; });
    bind attach.call to subscription DeliveryState.attach;
    bind status.get to subscription DeliveryState.status;
    bind retire.call to subscription DeliveryState.retire;
  }
}`,
    );
  const result = compileCapabilitySource(formatQx(ws), "subscriptions.qx", {
    interfaces,
    interfaceClosure: [...interfaces.values()],
  });
  assert.ok(result.ok, JSON.stringify(result.diagnostics));
  const kernel = compileKernelProgram(compileExecutionPlan(result.workspace)) as {
    model: {
      owners: { slots: { policy: string; subscription?: { operation: string; pins: unknown[]; active: unknown } }[] }[];
    };
  };
  const slot = kernel.model.owners.flatMap((owner) => owner.slots).find((slot) => slot.policy === "subscription")!;
  assert.equal(slot.subscription!.operation, "receive:call");
  assert.equal(slot.subscription!.pins.length, 2);
  assert.ok(slot.subscription!.active);
  const bad = compileCapabilitySource(ws.replace("on Delivery : string", "on Delivery : uint32"), "bad.qx", {
    interfaces,
    interfaceClosure: [...interfaces.values()],
  });
  assert.equal(bad.ok, false, "remembered parameter type must agree with attachment and handler");
});