Test file

Attempt history and isolation

What each message keeps as evidence: its signature check, its key, every attempt and its wait, a replay’s history, and attempts that can’t be rewritten.

tests/event-evidence.test.ts · 289 lines

import { afterEach, beforeEach, describe, it, expect } from "vitest";
import { randomUUID } from "node:crypto";
import { readFile } from "node:fs/promises";
import { PGlite } from "@electric-sql/pglite";
import type { FastifyInstance } from "fastify";
import { openStore, type Store } from "../packages/db/src/index";
import { createServer } from "../apps/api/src/server";
import { redactEventPayload } from "../packages/db/src/event-evidence";

type Detail = Awaited<ReturnType<Store["ops"]["detail"]>>;
type Session = Awaited<ReturnType<Store["createDemoSession"]>>;

const origin = "http://localhost:3000";

describe("stored event evidence, read through the API", () => {
  let clock: Date;
  let store: Store;
  let app: FastifyInstance;
  let demo: Session;
  let userId: string;

  beforeEach(async () => {
    clock = new Date("2026-10-13T12:00:00Z");
    store = await openStore({ now: () => clock });
    app = createServer(store, {
      origin,
      demoHistory: false,
      deliver: async () => {},
    });
    demo = await store.createDemoSession();
    userId = (await store.authenticate(demo.value)).id;
  });

  afterEach(async () => {
    await app.close();
    await store.close();
  });

  const cookie = () => `greenlot_session=${demo.value}`;
  const later = (ms: number) => {
    clock = new Date(clock.getTime() + ms);
  };
  const snapshot = () => store.ops.read(userId, demo.tenantId);

  async function simulate(action: string) {
    const response = await app.inject({
      method: "POST",
      url: `/api/tenants/${demo.tenantId}/simulate`,
      headers: { origin, cookie: cookie() },
      payload: { action },
    });
    expect(response.statusCode, response.body).toBe(200);
    await store.ops.tick();
    return snapshot();
  }

  async function detail(rawId: string): Promise<Detail> {
    const response = await app.inject({
      url: `/api/tenants/${demo.tenantId}/events/${rawId}`,
      headers: { cookie: cookie() },
    });
    expect(response.statusCode, response.body).toBe(200);
    return response.json();
  }

  it("records a repeat by its key, with no effects, pointing at the original", async () => {
    const state = await simulate("duplicate");
    const duplicate = state.events.find(
      (event) => event.job.status === "duplicate",
    )!;
    const repeat = await detail(duplicate.raw.id);

    expect(repeat.signature).toEqual({
      result: "accepted",
      algorithm: "HMAC-SHA256",
      header: "x-shopify-hmac-sha256",
    });
    const [decision] = repeat.decisions;
    expect(decision).toMatchObject({ duplicate: true, ordering: "duplicate" });
    expect(decision.key).toContain("order.");
    expect(repeat.effects).toEqual([]);

    const original = await detail(decision.originalRawId);
    const kinds = original.effects.map((effect) => effect.kind);
    expect(kinds).toContain("ledger");
    expect(kinds).toContain("outbound");
    expect(original.occurredAt).toBe(clock.toISOString());

    // The evidence names no workspace or connection.
    const json = JSON.stringify(repeat);
    expect(json).not.toContain(demo.tenantId);
    for (const connection of state.data.connections)
      expect(json).not.toContain(connection.id);
  });

  it("records an older version of an order as ignored, with no effects", async () => {
    const state = await simulate("out_of_order");
    const older = state.events.find((event) =>
      event.job.reason.includes("newer object version"),
    )!;
    const evidence = await detail(older.raw.id);

    expect(evidence.decisions[0].ordering).toBe("ignored_older");
    expect(evidence.effects).toEqual([]);
  });

  it("keeps every attempt of a transient failure and the wait it chose", async () => {
    const state = await simulate("transient");
    const retried = state.events.find(
      (event) => event.job.status === "retried",
    )!;
    const first = await detail(retried.raw.id);

    const [attempt] = first.attempts;
    expect(attempt.failureKind).toBe("transient");
    const due = new Date(first.nextAttemptAt!).getTime() - clock.getTime();
    expect(due).toBe(attempt.backoffMs);
    expect(attempt.effects).toEqual([]);

    // The first two waits are under ten seconds each.
    for (let wait = 0; wait < 2; wait++) {
      later(10000);
      await store.ops.tick();
    }
    const last = await detail(retried.raw.id);
    const outcomes = last.attempts.map((attempt) => attempt.outcome);
    expect(outcomes).toEqual(["retried", "retried", "processed"]);
    const ordinals = last.attempts.map((attempt) => attempt.ordinal);
    expect(ordinals).toEqual([1, 2, 3]);
    const kinds = last.effects.map((effect) => effect.kind);
    expect(kinds).toContain("ledger");
  });

  it("waits out a rate limit for as long as the provider asked", async () => {
    const state = await simulate("rate_limit");
    const limited = state.events.find(
      (event) => event.job.status === "retried",
    )!;
    const first = await detail(limited.raw.id);

    expect(first.attempts[0].failureKind).toBe("rate_limit");
    // The simulated provider sends Retry-After: 5.
    expect(first.attempts[0].backoffMs).toBe(5000);
    expect(first.signature.result).toBe("not_applicable");

    later(6000);
    await store.ops.tick();
    const last = await detail(limited.raw.id);
    expect(last.attempts.at(-1)?.outcome).toBe("processed");
  });

  it("adds a reviewed replay to the history and keeps the original bytes", async () => {
    const state = await simulate("invoice_error");
    const failed = state.events.find((event) => event.job.status === "failed")!;
    const bytes = failed.raw.body;

    await store.ops.command(userId, demo.tenantId, {
      type: "event.retry",
      id: failed.raw.id,
      note: "Reviewed simulator rejection before replay.",
    });
    await store.ops.tick();

    const replayed = await detail(failed.raw.id);
    const outcomes = replayed.attempts.map((attempt) => attempt.outcome);
    expect(outcomes).toEqual(["failed", "replay_queued", "processed"]);
    expect(replayed.attempts[1].kind).toBe("review");
    expect(replayed.canReplay).toBe(false);

    const after = await snapshot();
    const stored = after.events.find(
      (event) => event.raw.id === failed.raw.id,
    )!;
    expect(stored.raw.body).toBe(bytes);
    const failing = after.observability.providers.filter(
      (provider) => (provider.failureRate ?? 0) > 0,
    );
    expect(failing).not.toHaveLength(0);
  });

  it("answers 404 to another workspace and to an unknown event", async () => {
    const state = await simulate("duplicate");
    const eventUrl = (id: string) =>
      `/api/tenants/${demo.tenantId}/events/${id}`;
    const other = await store.createDemoSession();

    const fromOther = await app.inject({
      url: eventUrl(state.events[0].raw.id),
      headers: { cookie: `greenlot_session=${other.value}` },
    });
    expect(fromOther.statusCode).toBe(404);

    const unknown = await app.inject({
      url: eventUrl(randomUUID()),
      headers: { cookie: cookie() },
    });
    expect(unknown.statusCode).toBe(404);
  });
});

it("redacts free text, identities and secrets, keeping quantities and source times", () => {
  const output = redactEventPayload(
    JSON.stringify({
      id: "PRIVATE_ID",
      created_at: "2026-10-13T12:00:00Z",
      line_items: [
        {
          quantity: 2,
          variant_id: 1001,
          name: "PRIVATE_NAME",
          email: "PRIVATE_EMAIL",
        },
      ],
      access_token: "PRIVATE_SECRET",
      address: { street: "PRIVATE_ADDRESS" },
      nestedUnknown: { token: "PRIVATE_TOKEN" },
    }),
  );

  expect(JSON.stringify(output)).not.toContain("PRIVATE");
  expect(output).toMatchObject({
    created_at: "2026-10-13T12:00:00Z",
    line_items: [{ quantity: 2, variant_id: 1001 }],
  });
  expect(redactEventPayload("PRIVATE_NON_JSON")).toEqual({
    redacted: "Non-JSON payload omitted",
  });
});

it("keeps attempt records append-only, while an expiring workspace still takes them with it", async () => {
  const client = new PGlite();
  const tenant = randomUUID();
  const connection = randomUUID();
  const raw = randomUUID();
  const attempt = randomUUID();
  try {
    for (const file of [
      "0000_foundations.sql",
      "0001_operations.sql",
      "0012_event_attempts.sql",
    ]) {
      const migration = new URL(
        `../packages/db/migrations/${file}`,
        import.meta.url,
      );
      await client.exec(await readFile(migration, "utf8"));
    }
    await client.query(
      `INSERT INTO tenants (id, name) VALUES ($1, 'Synthetic event lab')`,
      [tenant],
    );
    await client.query(
      `INSERT INTO integration_connections
         (id, tenant_id, provider, secret, notification_url)
       VALUES ($1, $2, 'shopify', 'synthetic', 'http://localhost/webhook')`,
      [connection, tenant],
    );
    await client.query(
      `INSERT INTO raw_events
         (id, tenant_id, connection_id, body, headers, received_at,
          source, accepted)
       VALUES ($1, $2, $3, '{}', '{}', now(), 'webhook', true)`,
      [raw, tenant, connection],
    );
    await client.query(
      `INSERT INTO event_attempts
         (id, tenant_id, raw_id, ordinal, kind, started_at, finished_at,
          outcome)
       VALUES ($1, $2, $3, 1, 'worker', now(), now(), 'processed')`,
      [attempt, tenant, raw],
    );

    const rewrite = client.query(
      `UPDATE event_attempts SET outcome = 'failed' WHERE id = $1`,
      [attempt],
    );
    await expect(rewrite).rejects.toThrow("append-only");
    const erase = client.query(`DELETE FROM event_attempts WHERE id = $1`, [
      attempt,
    ]);
    await expect(erase).rejects.toThrow("append-only");

    await client.query(`DELETE FROM tenants WHERE id = $1`, [tenant]);
    const left = await client.query(`SELECT * FROM event_attempts`);
    expect(left.rows).toEqual([]);
  } finally {
    await client.close();
  }
});