jameskurz-filecoin · Oct 11, 2026 · Oct 11, 2026
@@ -0,0 +1,214 @@
+import { randomUUID } from "node:crypto";
+import { afterAll, beforeAll, expect, it, vi } from "vitest";
+import { provisionStaff } from "../../../scripts/provision-staff";
+import { createRuntimeDatabase } from "./client";
+import { accounts, organizations } from "./schema";
+
+const url =
+  process.env.DIRECT_DATABASE_URL ??
+  "postgresql://postgres:postgres@127.0.0.1:54322/postgres?sslmode=disable";
+const { db, client } = createRuntimeDatabase({
+  url,
+  role: "clockwork_service",
+  ssl: false,
+});
+const accountId = randomUUID();
+const organizationId = randomUUID();
+const workosOrganizationId = `org_${organizationId.replaceAll("-", "")}`;
+const operatorId = randomUUID();
+const historicalOrganizationEventId = randomUUID();
+const identities: string[] = [operatorId];
+const consoleLog = vi.spyOn(console, "log").mockImplementation(() => undefined);
+// Even an accidentally unmocked provider must never leave the local test.
+const fetchSpy = vi
+  .spyOn(globalThis, "fetch")
+  .mockRejectedValue(new Error("Unexpected provider call"));
+
+async function audit(
+  id: string,
+  aggregateType: string,
+  aggregateId: string,
+  version: number,
+) {
+  await client`insert into audit_events(id,aggregate_type,aggregate_id,aggregate_version,event_type,event_version,actor,occurred_at,request_id,after)
+    values(${id},${aggregateType},${aggregateId},${version},'test.staff.history',1,${JSON.stringify({ kind: "user", id: operatorId })}::jsonb,now(),${randomUUID()},'{}'::jsonb)`;
+}
+
+beforeAll(async () => {
+  await db.insert(accounts).values({
+    id: accountId,
+    legalName: `Provisioning ${accountId}`,
+    relationshipRoles: ["direct_client"],
+    registeredAddress: {
+      line1: "1 Main St",
+      city: "Wilmington",
+      region: "DE",
+      postalCode: "19801",
+      country: "US",
+    },
+    billingContact: { name: "Billing", email: "billing@fil.one" },
+    apContact: { name: "AP", email: "ap@fil.one" },
+    invoiceDeliveryEmail: "billing@fil.one",
+    domain: `provisioning-${accountId}.test`,
+    country: "US",
+    currency: "USD",
+  } as typeof accounts.$inferInsert);
+  await db.insert(organizations).values({
+    id: organizationId,
+    accountId,
+    name: "Provisioning audit",
+    side: "fil_one",
+    workosOrganizationId,
+  } as typeof organizations.$inferInsert);
+  await client`insert into commerce_users(id,workos_user_id,email,name,is_internal_staff,mfa_enrolled)
+    values(${operatorId},${`user_${operatorId}`},${`operator-${operatorId}@fil.one`},'Provisioning operator',true,true)`;
+  await client`insert into memberships(id,organization_id,user_id,role)
+    values(${randomUUID()},${organizationId},${operatorId},'commerce_admin')`;
+  await audit(historicalOrganizationEventId, "organization", organizationId, 1);
+});
+
+afterAll(async () => {
+  consoleLog.mockRestore();
+  fetchSpy.mockRestore();
+  await client`delete from memberships where organization_id=${organizationId}`;
+  // Retain identities referenced by immutable audit/notices, but remove staff eligibility.
+  await client`update commerce_users set is_internal_staff=false where id=any(${identities})`;
+  await client.end();
+});
+
+async function fixture(kind: "create" | "restore" | "update") {
+  const id = randomUUID();
+  const email = `provision-${id}@fil.one`;
+  const workosUserId = `user_${id.replaceAll("-", "")}`;
+  identities.push(id);
+  if (kind !== "create") {
+    await client`insert into commerce_users(id,workos_user_id,email,name,is_internal_staff,mfa_enrolled)
+      values(${id},${workosUserId},${email},'Existing staff',true,true)`;
+  }
+  const priorMembershipId = randomUUID();
+  if (kind !== "create") {
+    await client`insert into memberships(id,organization_id,user_id,role)
+      values(${priorMembershipId},${organizationId},${id},'internal_operator')`;
+    await audit(randomUUID(), "membership", priorMembershipId, 1);
+    if (kind === "restore")
+      await client`delete from memberships where id=${priorMembershipId}`;
+  }
+  const environment = {
+    STAFF_PROVISIONING: JSON.stringify({
+      email,
+      name: "Approved staff",
+      title: "Staff",
+      role: "commerce_admin",
+      updateRole: true,
+    }),
+    BOOTSTRAP_MANIFEST: JSON.stringify({
+      environment: "staging",
+      targetDatabaseHost: new URL(url).hostname,
+      operatorUserId: operatorId,
+      organization: { id: organizationId, workosOrganizationId },
+    }),
+    DIRECT_DATABASE_URL: url,
+    WORKOS_API_KEY: "test-provider-key",
+    DEPLOY_STAGE: "staging",
+    INTERNAL_EMAIL_DOMAINS: "fil.one",
+  };
+  const provider = vi.fn(() =>
+    Promise.resolve({
+      workosUserId,
+      workosMembershipId: `om_${id.replaceAll("-", "")}`,
+      outcome: "existing" as const,
+    }),
+  );
+  return { id, email, priorMembershipId, environment, provider };
+}
+
+async function membership(email: string) {
+  const [row] = await client<
+    { id: string; user_id: string; role: string; row_version: number }[]
+  >`
+    select m.id,m.user_id,m.role,m.row_version from memberships m join commerce_users u on u.id=m.user_id where u.email=${email}`;
+  if (!row) throw new Error("Expected provisioned membership");
+  return row;
+}
+
+it("updates the membership audit stream past existing organization and membership history", async () => {
+  const input = await fixture("update");
+  await provisionStaff(input.environment, input.provider);
+  expect(await membership(input.email)).toMatchObject({
+    id: input.priorMembershipId,
+    role: "commerce_admin",
+    row_version: 2,
+  });
+  input.environment.STAFF_PROVISIONING = JSON.stringify({
+    email: input.email,
+    name: "Approved staff",
+    title: "Staff",
+    role: "revenue",
+    updateRole: true,
+  });
+  await provisionStaff(input.environment, input.provider);
+  expect(await membership(input.email)).toMatchObject({
+    id: input.priorMembershipId,
+    role: "revenue",
+    row_version: 3,
+  });
+  // Reapplying the same approved request verifies without another mutation/audit.
+  await provisionStaff(input.environment, input.provider);
+  const events = await client<
+    {
+      aggregate_version: number;
+      event_type: string;
+      after: Record<string, unknown>;
+      outbox_count: number;
+    }[]
+  >`
+    select e.aggregate_version,e.event_type,e.after,(select count(*)::int from outbox_messages o where o.event_id=e.id) as outbox_count
+    from audit_events e where e.aggregate_type='membership' and e.aggregate_id=${input.priorMembershipId} order by e.aggregate_version`;
+  expect(events.map((event) => event.aggregate_version)).toEqual([1, 2, 3]);
+  expect(events.slice(1)).toMatchObject(
+    [2, 3].map((version) => ({
+      aggregate_version: version,
+      event_type: "staff.role_changed",
+      after: { organizationId },
+      outbox_count: 1,
+    })),
+  );
+  expect(fetchSpy).not.toHaveBeenCalled();
+  expect(input.provider).toHaveBeenCalledTimes(3);
+});
+
+it.each(["create", "restore"] as const)(
+  "%s uses a fresh membership stream despite existing organization history",
+  async (kind) => {
+    const input = await fixture(kind);
+    await provisionStaff(input.environment, input.provider);
+    const row = await membership(input.email);
+    identities.push(row.user_id);
+    expect(row).toMatchObject({ role: "commerce_admin", row_version: 1 });
+    expect(row.id).not.toBe(input.priorMembershipId);
+    const events = await client<
+      {
+        aggregate_version: number;
+        event_type: string;
+        after: Record<string, unknown>;
+        outbox_count: number;
+      }[]
+    >`
+    select e.aggregate_version,e.event_type,e.after,(select count(*)::int from outbox_messages o where o.event_id=e.id) as outbox_count
+    from audit_events e where e.aggregate_type='membership' and e.aggregate_id=${row.id}`;
+    expect(events).toMatchObject([
+      {
+        aggregate_version: 1,
+        event_type:
+          kind === "create" ? "staff.provisioned" : "staff.reactivated",
+        after: { organizationId },
+        outbox_count: 1,
+      },
+    ]);
+    const prior = await client<
+      { id: string }[]
+    >`select id from audit_events where aggregate_type='organization' and aggregate_id=${organizationId} and aggregate_version=1`;
+    expect(prior.map(({ id }) => id)).toEqual([historicalOrganizationEventId]);
+    expect(fetchSpy).not.toHaveBeenCalled();
+  },
+);
@@ -1,4 +1,6 @@
 import { randomUUID } from "node:crypto";
+import { realpathSync } from "node:fs";
+import { pathToFileURL } from "node:url";
 import {
   assertStaffAdministratorRemains,
   assertStaffProvisioningTarget,
@@ -13,24 +15,27 @@ import postgres from "postgres";
 import { provisionWorkosStaff } from "../packages/integrations/src/workos/provision-staff";
 
 let phase = "configuration";
-async function main() {
-  const raw = process.env.STAFF_PROVISIONING;
+export async function provisionStaff(
+  environment: NodeJS.ProcessEnv = process.env,
+  provisionIdentity: typeof provisionWorkosStaff = provisionWorkosStaff,
+) {
+  const raw = environment.STAFF_PROVISIONING;
   if (!raw) return;
   const person = StaffProvisioningSchema.parse(JSON.parse(raw));
   const manifest = JSON.parse(
-    process.env.BOOTSTRAP_MANIFEST ?? "null",
+    environment.BOOTSTRAP_MANIFEST ?? "null",
   ) as StaffProvisioningManifest | null;
-  const url = process.env.DIRECT_DATABASE_URL,
-    key = process.env.WORKOS_API_KEY;
+  const url = environment.DIRECT_DATABASE_URL,
+    key = environment.WORKOS_API_KEY;
   if (!manifest || !url || !key)
     throw new Error("STAFF_PROVISIONING_CONFIGURATION_REQUIRED");
   assertStaffProvisioningTarget({
     manifest,
     databaseHost: new URL(url).hostname,
-    deployStage: process.env.DEPLOY_STAGE,
+    deployStage: environment.DEPLOY_STAGE,
   });
   if (
-    !staffEmailDomainAllowed(person.email, process.env.INTERNAL_EMAIL_DOMAINS)
+    !staffEmailDomainAllowed(person.email, environment.INTERNAL_EMAIL_DOMAINS)
   )
     throw new Error("STAFF_PROVISIONING_DOMAIN_DENIED");
   const client = postgres(url, {
@@ -59,7 +64,7 @@ async function main() {
     });
     console.log("staff provisioning: identity plan", { plan: plan.kind });
     phase = "workos";
-    const binding = await provisionWorkosStaff(
+    const binding = await provisionIdentity(
       key,
       manifest.organization.workosOrganizationId,
       person,
@@ -74,13 +79,20 @@ async function main() {
     const actor = { kind: "user", id: manifest.operatorUserId };
     const appendEvent = async (
       tx: postgres.TransactionSql,
+      membership: { id: string; row_version: number },
       eventType: string,
       after: Record<string, string | boolean>,
       before?: Record<string, string | boolean>,
     ) => {
       const eventId = randomUUID();
-      await tx`insert into audit_events(id,aggregate_type,aggregate_id,aggregate_version,event_type,event_version,actor,occurred_at,request_id,before,after) values(${eventId},'organization',${manifest.organization.id},1,${eventType},1,${tx.json(actor)},now(),${randomUUID()},${before ? tx.json(before) : null},${tx.json(after)})`;
-      await tx`insert into outbox_messages(id,event_id,topic,payload) values(${randomUUID()},${eventId},${eventType},${tx.json(after)})`;
+      const associatedAfter = {
+        ...after,
+        organizationId: manifest.organization.id,
+      };
+      // Staff changes share Team's membership stream and its persisted version.
+      // Organization version 1 may already belong to an unrelated earlier event.
+      await tx`insert into audit_events(id,aggregate_type,aggregate_id,aggregate_version,event_type,event_version,actor,occurred_at,request_id,before,after) values(${eventId},'membership',${membership.id},${membership.row_version},${eventType},1,${tx.json(actor)},now(),${randomUUID()},${before ? tx.json(before) : null},${tx.json(associatedAfter)})`;
+      await tx`insert into outbox_messages(id,event_id,topic,payload) values(${randomUUID()},${eventId},${eventType},${tx.json(associatedAfter)})`;
     };
     if (plan.kind === "update_role") {
       phase = "update_role";
@@ -99,13 +111,16 @@ async function main() {
           otherAdministrators: administrators?.total ?? 0,
         });
         phase = "role_update";
-        const updated =
-          await tx`update memberships set role=${person.role} where user_id=${plan.userId} and organization_id=${manifest.organization.id} and role=${plan.from} returning id`;
-        if (updated.length !== 1)
+        const updated = await tx<
+          { id: string; row_version: number }[]
+        >`update memberships set role=${person.role} where user_id=${plan.userId} and organization_id=${manifest.organization.id} and role=${plan.from} returning id,row_version`;
+        const membership = updated[0];
+        if (updated.length !== 1 || !membership)
           throw new Error("STAFF_PROVISIONING_RETRY_REQUIRED");
         phase = "role_audit";
         await appendEvent(
           tx,
+          membership,
           "staff.role_changed",
           { email: person.email, role: person.role },
           { email: person.email, role: plan.from },
@@ -132,18 +147,30 @@ async function main() {
           await tx`select id from memberships where user_id=${plan.userId}`;
         if (current.length)
           throw new Error("STAFF_PROVISIONING_RETRY_REQUIRED");
-        await tx`insert into memberships(id,organization_id,user_id,workos_membership_id,role) values(${randomUUID()},${manifest.organization.id},${plan.userId},${binding.workosMembershipId},${person.role})`;
-        await appendEvent(tx, "staff.reactivated", after);
+        const membershipId = randomUUID();
+        await tx`insert into memberships(id,organization_id,user_id,workos_membership_id,role) values(${membershipId},${manifest.organization.id},${plan.userId},${binding.workosMembershipId},${person.role})`;
+        await appendEvent(
+          tx,
+          { id: membershipId, row_version: 1 },
+          "staff.reactivated",
+          after,
+        );
         return;
       }
       const duplicate =
         await tx`select id from commerce_users where lower(email)=${person.email}`;
       if (duplicate.length)
         throw new Error("STAFF_PROVISIONING_RETRY_REQUIRED");
       const id = randomUUID();
+      const membershipId = randomUUID();
       await tx`insert into commerce_users(id,workos_user_id,email,name,is_internal_staff,mfa_enrolled) values(${id},${binding.workosUserId},${person.email},${person.name},true,false)`;
-      await tx`insert into memberships(id,organization_id,user_id,workos_membership_id,role) values(${randomUUID()},${manifest.organization.id},${id},${binding.workosMembershipId},${person.role})`;
-      await appendEvent(tx, "staff.provisioned", after);
+      await tx`insert into memberships(id,organization_id,user_id,workos_membership_id,role) values(${membershipId},${manifest.organization.id},${id},${binding.workosMembershipId},${person.role})`;
+      await appendEvent(
+        tx,
+        { id: membershipId, row_version: 1 },
+        "staff.provisioned",
+        after,
+      );
     });
     console.log(
       `staff provisioning: identity and ${person.role} membership ${plan.kind === "restore" ? "restored" : "created"}; email verification and MFA remain required at login`,
@@ -152,10 +179,14 @@ async function main() {
     await client.end();
   }
 }
-main().catch((error: unknown) => {
-  console.error("staff provisioning failed", {
-    phase,
-    code: staffProvisioningFailureCode(error),
+if (
+  process.argv[1] &&
+  import.meta.url === pathToFileURL(realpathSync(process.argv[1])).href
+)
+  provisionStaff().catch((error: unknown) => {
+    console.error("staff provisioning failed", {
+      phase,
+      code: staffProvisioningFailureCode(error),
+    });
+    process.exitCode = 1;
   });
-  process.exitCode = 1;
-});