Test file

End-to-end worker behavior

The demo’s failures through the real API and worker: repeats, older versions, offline sales, outages, revoked access, retries and invoices, and who may do what.

tests/operations.test.ts · 480 lines

import { Pool } from "pg";
import { randomUUID, randomBytes } from "node:crypto";
import { beforeEach, afterEach, it, expect } from "vitest";
import type { FastifyInstance } from "fastify";
import { openStore, type Store } from "../packages/db/src/index";
import { createServer } from "../apps/api/src/server";

type Snapshot = Awaited<ReturnType<Store["ops"]["read"]>>;

const origin = "http://localhost:3000";
const encryptionKey = randomBytes(32);
// A new workspace opens with 30 bags of each product at the roastery.
const opening = 30;
// A sample workspace lasts a day.
const sampleLifetime = 24 * 60 * 60 * 1000;

let store: Store;
let app: FastifyInstance;
let clock: Date;
let cookie: string;
let tenantId: string;
let userId: string;
// Set when TEST_DATABASE_URL points the tests at a real PostgreSQL server.
let testPool: Pool | undefined;
let testSchema: string;

const later = (ms: number) => {
  clock = new Date(clock.getTime() + ms);
};
const post = (url: string, payload: object, session = cookie) =>
  app.inject({
    method: "POST",
    url,
    payload,
    headers: { origin, cookie: session },
  });
const snapshot = () => store.ops.read(userId, tenantId);
const stock = (state: Snapshot, sku = "espresso-250") =>
  state.inventory.find(
    (row) => row.sku === sku && row.locationId === "roastery",
  )!;
const selectAccounting = (provider: "xero" | "quickbooks") =>
  store.ops.command(userId, tenantId, { type: "accounting.select", provider });

async function simulate(action: string) {
  const response = await post(`/api/tenants/${tenantId}/simulate`, { action });
  expect(response.statusCode, response.body).toBe(200);
  await store.ops.tick();
  return snapshot();
}

beforeEach(async () => {
  clock = new Date("2026-10-11T10:00:00Z");
  let url: string | undefined;
  if (process.env.TEST_DATABASE_URL) {
    // Each test gets its own schema on that server, dropped afterwards.
    testPool = new Pool({ connectionString: process.env.TEST_DATABASE_URL });
    testSchema = "greenlot_test_" + randomUUID().replaceAll("-", "");
    await testPool.query("CREATE SCHEMA " + testSchema);
    const parsed = new URL(process.env.TEST_DATABASE_URL);
    parsed.searchParams.set("options", "-csearch_path=" + testSchema);
    url = parsed.toString();
  }
  store = await openStore({ url, now: () => clock, encryptionKey });
  app = createServer(store, {
    origin,
    demoHistory: false,
    deliver: async () => {},
  });
  const started = await post("/api/demo/start", {}, "");
  expect(started.statusCode, started.body).toBe(200);
  tenantId = started.json().tenantId;
  cookie = started.headers["set-cookie"]!.toString().split(";")[0];
  const me = await app.inject({ url: "/api/me", headers: { cookie } });
  userId = me.json().user.id;
});

afterEach(async () => {
  await app.close();
  await store.close();
  if (testPool) {
    // The name is interpolated into DROP SCHEMA, so check it is ours.
    if (!/^greenlot_test_[a-f0-9]{32}$/.test(testSchema))
      throw new Error("Unexpected test schema");
    await testPool.query("DROP SCHEMA " + testSchema + " CASCADE");
    await testPool.end();
    testPool = undefined;
  }
});

it("deduplicates signed deliveries and ignores older snapshots", async () => {
  let state = await simulate("duplicate");
  expect(state.data.domain.sales).toHaveLength(1);
  const repeats = state.events.filter(
    (event) => event.job.status === "duplicate",
  );
  expect(repeats).toHaveLength(1);
  expect(stock(state).available).toBe(opening - 1);

  state = await simulate("out_of_order");
  expect(stock(state).available).toBe(opening - 3);
  const ignored = state.events.filter((event) =>
    event.job.reason.includes("newer object version"),
  );
  expect(ignored).not.toHaveLength(0);
});

it("opens one offline oversell in the order sales happened and resolves it by an audited transfer", async () => {
  let state = await simulate("offline_oversell");
  expect(state.data.domain.incidents).toHaveLength(1);
  const [incident] = state.data.domain.incidents;
  expect(incident.shortage).toBe(1);
  const [first, second] = incident.sales;
  expect(first.source).toBe("square");
  expect(first.occurredAt < second.occurredAt).toBe(true);

  await store.ops.command(userId, tenantId, {
    type: "incident.resolve",
    id: incident.id,
    action: "transfer",
    fromLocation: "cafe",
    note: "Transfer one bag to fulfill earlier offline sale.",
  });
  await store.ops.tick();

  state = await snapshot();
  expect(state.data.domain.incidents[0].status).toBe("resolved");
  expect(stock(state).available).toBe(0);
});

it("fetches the orders missed while a provider was offline", async () => {
  let state = await simulate("outage");
  expect(state.data.domain.sales).toHaveLength(0);

  // The simulated outage lasts ten seconds.
  later(11000);
  await store.ops.tick();
  state = await snapshot();
  expect(state.data.domain.sales).toHaveLength(2);
  const fetched = state.events.filter((event) => event.raw.source === "poll");
  expect(fetched).not.toHaveLength(0);

  // Fetching again finds nothing new.
  await store.ops.tick();
  expect((await snapshot()).data.domain.sales).toHaveLength(2);
});

it("pauses on a revoked token, then reconnects and replays safely", async () => {
  let state = await simulate("token_expiry");
  const paused = state.events.filter((event) => event.job.status === "paused");
  expect(paused).not.toHaveLength(0);
  expect(state.data.domain.sales).toHaveLength(0);
  const square = state.data.connections.find(
    (connection) => connection.provider === "square",
  )!;

  await store.ops.command(userId, tenantId, {
    type: "connection.reconnect",
    id: square.id,
  });
  await store.ops.tick();

  state = await snapshot();
  expect(state.data.domain.sales).toHaveLength(1);
  const reconnected = state.data.connections.find(
    (connection) => connection.provider === "square",
  )!;
  expect(reconnected.status).toBe("healthy");
});

it("keeps transient retries and rate-limit waits, then recovers without selling twice", async () => {
  let state = await simulate("transient");
  const retried = state.events.filter(
    (event) => event.job.status === "retried",
  );
  expect(retried).not.toHaveLength(0);
  for (let wait = 0; wait < 3; wait++) {
    later(10000);
    await store.ops.tick();
  }
  state = await snapshot();
  expect(state.data.domain.sales).toHaveLength(1);

  state = await simulate("rate_limit");
  const waiting = state.events.filter(
    (event) => event.job.status === "retried",
  );
  expect(waiting).not.toHaveLength(0);
  // The simulated provider sends Retry-After: 5.
  later(6000);
  await store.ops.tick();
  expect((await snapshot()).data.domain.sales).toHaveLength(2);
});

it("a newer stock count takes the place and the wait of the one it replaces", async () => {
  // Shopify is rate-limited for five seconds, so the sale's stock count for it waits.
  let state = await simulate("rate_limit");
  const waiting = state.events.find(
    (event) =>
      event.raw.source === "outbound" && event.job.status === "retried",
  )!;
  expect(waiting.job.coalesceKey).toBeTruthy();
  // A second sale changes the same count during the wait.
  later(1000);
  state = await simulate("sale");
  const same = state.events.filter(
    (event) => event.job.coalesceKey === waiting.job.coalesceKey,
  );
  expect(same.find((event) => event.raw.id === waiting.raw.id)!.job.status).toBe(
    "coalesced",
  );
  const next = same.find((event) => event.job.status !== "coalesced")!;
  // It waits where the replaced count waited: not sent into the limit, and not sent to the back.
  expect(next.job.status).toBe("queued");
  expect(next.job.attempts).toBe(0);
  expect(new Date(next.job.availableAt).getTime()).toBe(
    new Date(waiting.job.availableAt).getTime(),
  );
  later(5000);
  await store.ops.tick();
  state = await snapshot();
  const sent = state.events.find((event) => event.raw.id === next.raw.id)!;
  expect(sent.job.status).toBe("processed");
  expect(sent.job.operation).toMatchObject({
    quantity: stock(state).available,
  });
});

it.each(["xero", "quickbooks"] as const)(
  "%s holds a failed invoice in credit exposure, then takes an explicit retry and a payment",
  async (provider) => {
    await selectAccounting(provider);
    let state = await simulate("invoice_error");
    expect(state.data.wholesale.invoices[0].status).toBe("PENDING");
    const failed = state.events.find((event) => event.job.status === "failed")!;
    expect(failed.job.reason).toContain("validation");

    await store.ops.command(userId, tenantId, {
      type: "event.retry",
      id: failed.raw.id,
      note: "Corrected demo account code.",
    });
    await store.ops.tick();
    state = await snapshot();
    const [invoice] = state.data.wholesale.invoices;
    expect(invoice.externalId).toBeTruthy();
    // Xero's status name, which the demo uses for both providers.
    expect(invoice.status).toBe("AUTHORISED");

    state = await simulate("payment");
    const [paid] = state.data.wholesale.invoices;
    expect(paid.paidCents).toBe(2400);
    expect(paid.status).toBe("PAID");
  },
);

it("rejects bad signatures and enforces origin, workspace scope and expiry", async () => {
  const [connection] = (await snapshot()).data.connections;
  const unsigned = await app.inject({
    method: "POST",
    url: `/webhooks/shopify/${connection.id}`,
    headers: { "content-type": "application/json" },
    payload: '{"id":1}',
  });
  expect(unsigned.statusCode).toBe(401);
  // Refused, and still kept as received.
  expect((await snapshot()).events[0].job.status).toBe("failed");

  const other = await store.createDemoSession();
  const otherUser = await store.authenticate(other.value);
  await expect(store.ops.read(otherUser.id, tenantId)).rejects.toThrow(
    "Workspace not found",
  );

  const foreign = await app.inject({
    method: "POST",
    url: "/api/demo/start",
    headers: { origin: "https://other.example" },
    payload: {},
  });
  expect(foreign.statusCode).toBe(403);

  later(sampleLifetime + 1);
  await store.ops.tick();
  const token = cookie.split("=")[1];
  await expect(store.authenticate(token)).rejects.toThrow();
  await expect(store.ops.read(userId, tenantId)).rejects.toThrow(
    "Workspace not found",
  );
});

it("keeps new sample workspaces apart and detects inventory drift", async () => {
  await simulate("sale");
  const other = await store.createDemoSession();
  const otherUser = await store.authenticate(other.value);
  const theirs = await store.ops.read(otherUser.id, other.tenantId);
  expect(theirs.data.domain.sales).toHaveLength(0);

  const state = await simulate("drift");
  const open = state.data.domain.drifts.filter(
    (drift) => drift.status === "open",
  );
  expect(open).not.toHaveLength(0);
  expect(JSON.stringify(state)).not.toContain("demo-only");
});

it("applies a signed refund once, and a later worker pass doesn't apply it again", async () => {
  await simulate("sale");
  let state = await simulate("refund");
  expect(stock(state).onHand).toBe(opening);
  expect(state.data.domain.refunds).toHaveLength(1);

  await store.ops.tick();
  state = await snapshot();
  expect(state.data.domain.refunds).toHaveLength(1);
});

it("lets staff inspect operations but not change or simulate them", async () => {
  const link = await store.issueLink("staff-ops@example.test");
  const session = await store.consumeLink(link);
  const invite = await store.invite(
    userId,
    tenantId,
    "staff-ops@example.test",
    "staff",
  );
  await store.accept(session.user.id, session.user.email, invite.value);
  const staff = session.user.id;

  expect((await store.ops.read(staff, tenantId)).role).toBe("staff");
  await expect(
    store.ops.command(staff, tenantId, {
      type: "wholesale.create",
      accountId: "account-1",
      sku: "filter-bulk",
      quantity: 1,
    }),
  ).rejects.toThrow("manager");
  await expect(store.ops.simulate(staff, tenantId, "sale")).rejects.toThrow(
    "manager",
  );
});

it("refuses to retry an invalid signature or touch another workspace's event", async () => {
  const [connection] = (await snapshot()).data.connections;
  const unsigned = await store.ops.ingest(connection.id, {
    body: "{}",
    headers: {},
    notificationUrl: "",
  });
  await expect(
    store.ops.command(userId, tenantId, {
      type: "event.retry",
      id: unsigned.id,
      note: "Try again",
    }),
  ).rejects.toThrow("Invalid signatures");

  const other = await store.createDemoSession();
  const otherUser = await store.authenticate(other.value);
  await expect(
    store.ops.command(otherUser.id, other.tenantId, {
      type: "event.ignore",
      id: unsigned.id,
      note: "Ignore",
    }),
  ).rejects.toThrow("Event not found");
});

it("selects exactly one demo accounting provider, idempotently, and freezes it once it has history", async () => {
  const chosen = await post(`/api/tenants/${tenantId}/commands`, {
    type: "accounting.select",
    provider: "quickbooks",
  });
  expect(chosen.statusCode).toBe(200);
  const accounting = (state: Snapshot) =>
    state.data.connections
      .map((connection) => connection.provider)
      .filter((provider) => provider === "quickbooks" || provider === "xero");
  expect(accounting(await snapshot())).toEqual(["quickbooks"]);

  await selectAccounting("quickbooks");
  await selectAccounting("xero");
  await simulate("invoice_error");
  await expect(selectAccounting("quickbooks")).rejects.toThrow("history");

  const state = await snapshot();
  expect(accounting(state)).toContain("xero");
  const audited = state.data.audit.filter(
    (entry) => entry.action === "accounting.select",
  );
  expect(audited).not.toHaveLength(0);
});

it("refuses an accounting choice from a non-member, a manager, or a workspace without demo providers", async () => {
  const session = await store.consumeLink(
    await store.issueLink("accounting-manager@example.test"),
  );
  const manager = session.user.id;
  const choose = (workspace: string) =>
    store.ops.command(manager, workspace, {
      type: "accounting.select",
      provider: "quickbooks",
    });

  // Not a member yet: the workspace doesn't exist for them.
  await expect(choose(tenantId)).rejects.toMatchObject({ status: 404 });
  const invitation = await store.invite(
    userId,
    tenantId,
    session.user.email,
    "manager",
  );
  await store.accept(manager, session.user.email, invitation.value);
  await expect(choose(tenantId)).rejects.toMatchObject({ status: 403 });

  const normal = await store.createTenant(manager, "No provider accounts");
  await expect(choose(normal.id)).rejects.toThrow("not configured");
});

it("serializes concurrent provider choices into one accounting connection", async () => {
  await Promise.all([selectAccounting("quickbooks"), selectAccounting("xero")]);
  const accounting = (await snapshot()).data.connections.filter(
    (connection) =>
      connection.provider === "xero" || connection.provider === "quickbooks",
  );
  expect(accounting).toHaveLength(1);
});

it("serializes a provider choice against a dispatch, so its invoice goes to one provider", async () => {
  await store.ops.command(userId, tenantId, {
    type: "wholesale.create",
    accountId: "account-1",
    sku: "filter-bulk",
    quantity: 1,
  });
  const [order] = (await snapshot()).data.wholesale.orders;

  const results = await Promise.allSettled([
    selectAccounting("quickbooks"),
    store.ops.command(userId, tenantId, {
      type: "wholesale.dispatch",
      id: order.id,
    }),
  ]);
  expect(results[1].status).toBe("fulfilled");
  await store.ops.tick();

  const state = await snapshot();
  const { provider } = state.data.connections.find(
    (connection) =>
      connection.provider === "xero" || connection.provider === "quickbooks",
  )!;
  expect(state.data.wholesale.invoices).toHaveLength(1);
  expect(state.data.wholesale.invoices[0].externalId).toBeTruthy();
  const sent = state.events.filter(
    (event) => event.job.eventType === "invoice.create",
  );
  expect(sent).toHaveLength(1);
  expect(sent[0].provider).toBe(provider);
  expect(stock(state, "filter-bulk").onHand).toBe(opening - 1);
});

it("keeps even a rejected accounting receipt and won't reread it after a provider switch", async () => {
  const xero = (await snapshot()).data.connections.find(
    (connection) => connection.provider === "xero",
  )!;
  await app.inject({
    method: "POST",
    url: `/webhooks/xero/${xero.id}`,
    headers: { "content-type": "application/json" },
    payload: "{}",
  });
  await expect(selectAccounting("quickbooks")).rejects.toThrow("history");

  const [receipt] = (await snapshot()).events;
  expect(receipt.provider).toBe("xero");
  expect(receipt.raw.body).toBe("{}");
  expect(receipt.raw.accepted).toBe(false);
});