fix: make payment event reduction monotonic

packages/payment/README.md

@@ -3,3 +3,5 @@
 Gateway-neutral, webhook-authoritative payments for WrNexus. Amounts are integer minor units. Every initialize and refund operation requires an idempotency key. The package never accepts or stores card numbers, CVVs, or expiry input; its components navigate to provider-hosted checkout.
 
 Built-in adapters: offline `sandbox`, Stripe, Razorpay, and PayPal. Applications may add adapters with `defineGateway`. Capabilities are explicit, unsupported operations are refused, signed webhooks are mandatory in production, and package-owned migrations provision the append-only event ledger.
+
+Webhook events are deduplicated by the provider's event ID and reduced in provider occurrence/sequence order. Captured and refund states are monotonic, so a delayed retry of an older event cannot regress a terminal payment.

packages/payment/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@wrnexus/payment",
-  "version": "0.8.0",
+  "version": "0.8.1",
   "private": true,
   "type": "module",
   "main": "src/index.ts",

packages/payment/src/adapters.ts

@@ -131,6 +131,8 @@ export function sandboxGateway(options: SandboxOptions = {}): PaymentGateway {
         intentRef: String(payload.intentRef),
         type: String(payload.type ?? "payment.updated"),
         status: payload.status,
+        occurredAt: Number(payload.occurredAt ?? Date.now()),
+        sequence: payload.sequence,
         payload,
       };
     },
@@ -284,6 +286,7 @@ export function stripeGateway(config: StripeConfig): PaymentGateway<StripeConfig
         intentRef: object.payment_intent ?? object.id,
         type: row.type,
         status: statusOf(object.status),
+        occurredAt: Number(row.created ?? timestamp) * 1000,
         payload: row,
       };
     },
@@ -413,6 +416,8 @@ export function razorpayGateway(config: RazorpayConfig): PaymentGateway<Razorpay
         intentRef: payment.order_id ?? payment.id,
         type: row.event,
         status: statusOf(payment.status),
+        occurredAt:
+          Number(payment.created_at ?? row.created_at ?? Math.floor(Date.now() / 1000)) * 1000,
         payload: row,
       };
     },
@@ -540,6 +545,7 @@ export function paypalGateway(config: PayPalConfig): PaymentGateway<PayPalConfig
         intentRef: row.resource?.supplementary_data?.related_ids?.order_id ?? row.resource?.id,
         type: row.event_type,
         status: statusOf(String(row.resource?.status ?? "").toLowerCase()),
+        occurredAt: Date.parse(row.create_time ?? "") || Date.now(),
         payload: row,
       };
     },

packages/payment/src/contract.ts

@@ -1,4 +1,4 @@
-import type { PaymentGateway } from "./types.ts";
+import type { CreateIntentInput, GatewayIntent, PaymentGateway } from "./types.ts";
 
 export interface GatewayContractReport {
   ok: boolean;
@@ -36,3 +36,83 @@ export function verifyGatewayContract(adapter: PaymentGateway): GatewayContractR
     issues.push(`${adapter.id}: signed webhooks are required`);
   return { ok: issues.length === 0, issues };
 }
+
+export interface GatewayContractFixture {
+  input: CreateIntentInput;
+  webhook(intent: GatewayIntent): Request | Promise<Request>;
+}
+
+/** Executable lifecycle contract shared by built-in and third-party adapters. */
+export async function exerciseGatewayContract(
+  adapter: PaymentGateway,
+  fixture: GatewayContractFixture,
+): Promise<GatewayContractReport> {
+  const report = verifyGatewayContract(adapter);
+  const issues = [...report.issues];
+  const check = (condition: unknown, message: string) => {
+    if (!condition) issues.push(`${adapter.id}: ${message}`);
+  };
+  try {
+    const intent = await adapter.createIntent(fixture.input);
+    check(Boolean(intent.ref), "createIntent returned no reference");
+    check(intent.amount === fixture.input.amount, "createIntent changed the amount");
+    check(intent.currency === fixture.input.currency, "createIntent changed the currency");
+    const duplicate = await adapter.createIntent(fixture.input);
+    check(duplicate.ref === intent.ref, "createIntent did not honor its idempotency key");
+    const fetched = await adapter.fetchIntent(intent.ref);
+    check(fetched.ref === intent.ref, "fetchIntent returned a different reference");
+    const client = adapter.clientConfig(intent);
+    const configuredSecrets = Object.entries((adapter.config ?? {}) as Record<string, unknown>)
+      .filter(([key, value]) => /secret|password/i.test(key) && typeof value === "string")
+      .map(([, value]) => value);
+    check(
+      !configuredSecrets.some((secret) => Object.values(client).includes(secret as string)),
+      "clientConfig exposed a configured secret",
+    );
+    if (adapter.capabilities.authorizeThenCapture) {
+      const captured = await adapter.capture!(intent.ref);
+      check(Boolean(captured.ref), "capture returned no reference");
+    }
+    const cancellable = await adapter.createIntent({
+      ...fixture.input,
+      id: `${fixture.input.id}-cancel`,
+      idempotencyKey: `${fixture.input.idempotencyKey}:cancel`,
+    });
+    const cancelled = await adapter.cancelIntent(cancellable.ref, "contract");
+    check(cancelled.status === "cancelled", "cancelIntent did not return cancelled state");
+    const refund = await adapter.refund({
+      intentRef: intent.ref,
+      amount: Math.max(1, Math.floor(intent.amount / 2)),
+      currency: intent.currency,
+      reason: "contract",
+      idempotencyKey: `${fixture.input.idempotencyKey}:refund`,
+    });
+    check(Boolean(refund.ref), "refund returned no reference");
+    if (adapter.capabilities.customerVault) {
+      const customer = await adapter.createCustomer!({ subject: fixture.input.subject });
+      check(Boolean(customer), "createCustomer returned no reference");
+    }
+    if (adapter.capabilities.storedMethods) {
+      const customer = "contract-customer";
+      const method = await adapter.attachMethod!(customer, "contract-token-4242");
+      check(Boolean(method.ref), "attachMethod returned no reference");
+      check(
+        Array.isArray(await adapter.listMethods!(customer)),
+        "listMethods did not return an array",
+      );
+      await adapter.detachMethod!(method.ref);
+    }
+    const event = await adapter.verifyWebhook(
+      await fixture.webhook(intent),
+      "contract-webhook-secret",
+    );
+    check(Boolean(event.id), "verified webhook has no gateway event id");
+    check(Boolean(event.intentRef), "verified webhook has no intent reference");
+    check(Number.isFinite(event.occurredAt), "verified webhook has no gateway occurrence time");
+  } catch (error) {
+    issues.push(
+      `${adapter.id}: lifecycle threw ${error instanceof Error ? error.message : String(error)}`,
+    );
+  }
+  return { ok: issues.length === 0, issues };
+}

packages/payment/src/core.ts

@@ -39,7 +39,35 @@ const gatewayFault = <T>(reason: string): PaymentResult<T> => ({
   reason,
 });
 export function derivePaymentStatus(events: readonly PaymentEvent[]): PaymentStatus {
-  return events.at(-1)?.status ?? "created";
+  const ordered = [...events].sort((left, right) => {
+    if (left.sequence !== undefined && right.sequence !== undefined) {
+      const numeric = Number(left.sequence) - Number(right.sequence);
+      if (Number.isFinite(numeric) && numeric !== 0) return numeric;
+      const lexical = String(left.sequence).localeCompare(String(right.sequence));
+      if (lexical) return lexical;
+    }
+    return (
+      (left.occurredAt ?? left.at) - (right.occurredAt ?? right.at) ||
+      left.id.localeCompare(right.id)
+    );
+  });
+  let status: PaymentStatus = "created";
+  for (const event of ordered) {
+    if (status === "refunded") continue;
+    if (
+      status === "partially_refunded" &&
+      !["partially_refunded", "refunded"].includes(event.status)
+    )
+      continue;
+    if (
+      status === "captured" &&
+      ["created", "pending", "authorized", "failed", "cancelled"].includes(event.status)
+    )
+      continue;
+    if (status === "authorized" && ["created", "pending"].includes(event.status)) continue;
+    status = event.status;
+  }
+  return status;
 }
 export function defineGateway<T extends PaymentGateway>(gateway: T): T {
   if (!gateway.id.trim()) throw new Error("WRN-PAYMENT-GATEWAY: id is required");
@@ -269,6 +297,8 @@ export function definePayment(options: PaymentOptions) {
       type: event.type,
       status: event.status,
       at: Date.now(),
+      occurredAt: event.occurredAt,
+      sequence: event.sequence,
       payload: event.payload,
     });
   }

packages/payment/src/database.ts

@@ -84,8 +84,17 @@ export function databasePaymentStore(database = "default"): PaymentStore {
     async appendEvent(row) {
       try {
         const result = await db().exec(
-          "INSERT INTO wrn_payment_event (id,intent_id,type,status,created_at,payload_json) VALUES (?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING",
-          [row.id, row.intentId, row.type, row.status, row.at, JSON.stringify(row.payload ?? null)],
+          "INSERT INTO wrn_payment_event (id,intent_id,type,status,created_at,occurred_at,gateway_sequence,payload_json) VALUES (?,?,?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING",
+          [
+            row.id,
+            row.intentId,
+            row.type,
+            row.status,
+            row.at,
+            row.occurredAt ?? row.at,
+            row.sequence === undefined ? null : String(row.sequence),
+            JSON.stringify(row.payload ?? null),
+          ],
         );
         return result.changes > 0;
       } catch {
@@ -95,7 +104,7 @@ export function databasePaymentStore(database = "default"): PaymentStore {
     async events(id) {
       return (
         await db().all<any>(
-          "SELECT * FROM wrn_payment_event WHERE intent_id = ? ORDER BY created_at ASC, id ASC",
+          "SELECT * FROM wrn_payment_event WHERE intent_id = ? ORDER BY occurred_at ASC, gateway_sequence ASC, id ASC",
           [id],
         )
       ).map((row) => ({
@@ -104,6 +113,8 @@ export function databasePaymentStore(database = "default"): PaymentStore {
         type: String(row.type),
         status: row.status,
         at: Number(row.created_at),
+        occurredAt: Number(row.occurred_at ?? row.created_at),
+        sequence: row.gateway_sequence == null ? undefined : String(row.gateway_sequence),
         payload: parse(row.payload_json, null),
       }));
     },

packages/payment/src/plugin.ts

@@ -19,12 +19,19 @@ DROP TABLE IF EXISTS wrn_payment_method;
 DROP TABLE IF EXISTS wrn_payment_refund;
 DROP TABLE IF EXISTS wrn_payment_event;
 DROP TABLE IF EXISTS wrn_payment_intent;`;
+const orderingMigration = `-- +up
+ALTER TABLE wrn_payment_event ADD COLUMN occurred_at BIGINT;
+ALTER TABLE wrn_payment_event ADD COLUMN gateway_sequence TEXT;
+UPDATE wrn_payment_event SET occurred_at = created_at WHERE occurred_at IS NULL;
+CREATE INDEX IF NOT EXISTS wrn_payment_event_gateway_order ON wrn_payment_event(intent_id, occurred_at, gateway_sequence, id);
+-- +down
+DROP INDEX IF EXISTS wrn_payment_event_gateway_order;`;
 
 export function paymentPlugin() {
   const key = "@wrnexus/payment:enabled";
   return definePlugin({
     name: "@wrnexus/payment",
-    version: "0.8.0",
+    version: "0.8.1",
     componentDirs: [join(root, "components")],
     routeEntries(context) {
       return context.metadata.get(key)
@@ -45,7 +52,12 @@ export function paymentPlugin() {
         throw new Error("WRN-PAYMENT-CONFIG: sandbox cannot be enabled in production");
     },
     migrations(context: PluginContext) {
-      return context.metadata.get(key) ? [{ id: "wrnexus-payment-001", source: migration }] : [];
+      return context.metadata.get(key)
+        ? [
+            { id: "wrnexus-payment-001", source: migration },
+            { id: "wrnexus-payment-002-event-order", source: orderingMigration },
+          ]
+        : [];
     },
     documentation: [join(root, "README.md")],
   });

packages/payment/src/types.ts

@@ -73,6 +73,10 @@ export interface WebhookEvent {
   intentRef: string;
   type: string;
   status: PaymentStatus;
+  /** Gateway-authored event time, never local arrival time. */
+  occurredAt: number;
+  /** Gateway-authored ordering token when the provider exposes one. */
+  sequence?: string | number;
   payload: unknown;
 }
 
@@ -111,6 +115,8 @@ export interface PaymentEvent {
   type: string;
   status: PaymentStatus;
   at: number;
+  occurredAt?: number;
+  sequence?: string | number;
   payload?: unknown;
 }
 export interface PaymentRefund {

packages/payment/test/adapter-contract.test.ts

@@ -0,0 +1,237 @@
+import { afterEach, expect, test } from "bun:test";
+import {
+  exerciseGatewayContract,
+  paypalGateway,
+  razorpayGateway,
+  sandboxGateway,
+  signSandboxWebhook,
+  stripeGateway,
+  type GatewayIntent,
+} from "../src/index.ts";
+
+const originalFetch = globalThis.fetch;
+afterEach(() => {
+  globalThis.fetch = originalFetch;
+});
+const json = (value: unknown, status = 200) =>
+  new Response(JSON.stringify(value), { status, headers: { "content-type": "application/json" } });
+const encoder = new TextEncoder();
+async function signature(body: string, secret: string) {
+  const key = await crypto.subtle.importKey(
+    "raw",
+    encoder.encode(secret),
+    { name: "HMAC", hash: "SHA-256" },
+    false,
+    ["sign"],
+  );
+  return [...new Uint8Array(await crypto.subtle.sign("HMAC", key, encoder.encode(body)))]
+    .map((x) => x.toString(16).padStart(2, "0"))
+    .join("");
+}
+const input = {
+  id: "contract-pay",
+  subject: "contract-user",
+  amount: 1000,
+  currency: "USD",
+  idempotencyKey: "contract-init",
+};
+
+test("shared executable contract drives sandbox", async () => {
+  const adapter = sandboxGateway();
+  const report = await exerciseGatewayContract(adapter, {
+    input,
+    async webhook(intent) {
+      const signed = await signSandboxWebhook(
+        {
+          id: "sb-event",
+          intentRef: intent.ref,
+          type: "payment.captured",
+          status: "captured",
+          occurredAt: 100,
+        },
+        "contract-webhook-secret",
+      );
+      return new Request("https://test", {
+        method: "POST",
+        headers: { "x-wrnexus-signature": signed.signature },
+        body: signed.body,
+      });
+    },
+  });
+  expect(report).toEqual({ ok: true, issues: [] });
+});
+
+test("shared executable contract drives Stripe through its HTTP adapter", async () => {
+  globalThis.fetch = (async (value, init) => {
+    const url = String(value),
+      method = init?.method ?? "GET";
+    if (
+      url.includes("/payment_intents") &&
+      method === "POST" &&
+      !url.endsWith("/capture") &&
+      !url.endsWith("/cancel")
+    ) {
+      const body = String(init?.body);
+      const id = body.includes("cancel") ? "pi_cancel" : "pi_contract";
+      return json({
+        id,
+        status: "requires_capture",
+        amount: 1000,
+        amount_received: 1000,
+        currency: "usd",
+        client_secret: "pi_client_safe",
+      });
+    }
+    if (url.endsWith("/capture"))
+      return json({
+        id: "pi_contract",
+        status: "succeeded",
+        amount_received: 1000,
+        currency: "usd",
+      });
+    if (url.endsWith("/cancel"))
+      return json({ id: "pi_cancel", status: "canceled", amount: 1000, currency: "usd" });
+    if (url.includes("/payment_intents/"))
+      return json({ id: "pi_contract", status: "requires_capture", amount: 1000, currency: "usd" });
+    if (url.endsWith("/refunds")) return json({ id: "re_1", status: "succeeded", amount: 500 });
+    if (url.endsWith("/customers")) return json({ id: "cus_1" });
+    if (url.includes("/payment_methods?")) return json({ data: [] });
+    if (url.includes("/payment_methods/") && url.endsWith("/attach"))
+      return json({
+        id: "pm_1",
+        card: { brand: "visa", last4: "4242", exp_month: 1, exp_year: 2030 },
+      });
+    if (url.includes("/payment_methods/") && url.endsWith("/detach")) return json({ id: "pm_1" });
+    return json({}, 404);
+  }) as typeof fetch;
+  const adapter = stripeGateway({
+    secretKey: "sk_test_secret",
+    publishableKey: "pk_test",
+    webhookSecret: "contract-webhook-secret",
+  });
+  const report = await exerciseGatewayContract(adapter, {
+    input,
+    async webhook(intent) {
+      const body = JSON.stringify({
+        id: "evt_1",
+        created: 100,
+        data: { object: { id: intent.ref, status: "succeeded" } },
+        type: "payment_intent.succeeded",
+      });
+      return new Request("https://test", {
+        method: "POST",
+        headers: {
+          "stripe-signature": `t=100,v1=${await signature(`100.${body}`, "contract-webhook-secret")}`,
+        },
+        body,
+      });
+    },
+  });
+  expect(report).toEqual({ ok: true, issues: [] });
+});
+
+test("shared executable contract drives Razorpay through its HTTP adapter", async () => {
+  globalThis.fetch = (async (value, init) => {
+    const url = String(value),
+      method = init?.method ?? "GET";
+    if (url.endsWith("/orders") && method === "POST") {
+      const body = String(init?.body);
+      return json({
+        id: body.includes("cancel") ? "order_cancel" : "order_1",
+        status: "created",
+        amount: 1000,
+        currency: "USD",
+      });
+    }
+    if (url.includes("/orders/"))
+      return json({
+        id: url.endsWith("order_cancel") ? "order_cancel" : "order_1",
+        status: "created",
+        amount: 1000,
+        currency: "USD",
+      });
+    if (url.includes("/refund")) return json({ id: "rfnd_1", status: "processed", amount: 500 });
+    return json({}, 404);
+  }) as typeof fetch;
+  const adapter = razorpayGateway({
+    keyId: "rzp_test",
+    keySecret: "key_secret",
+    webhookSecret: "contract-webhook-secret",
+  });
+  const report = await exerciseGatewayContract(adapter, {
+    input,
+    async webhook(intent) {
+      const body = JSON.stringify({
+        id: "rz_evt",
+        event: "payment.captured",
+        payload: {
+          payment: {
+            entity: { id: "pay_1", order_id: intent.ref, status: "captured", created_at: 100 },
+          },
+        },
+      });
+      return new Request("https://test", {
+        method: "POST",
+        headers: { "x-razorpay-signature": await signature(body, "contract-webhook-secret") },
+        body,
+      });
+    },
+  });
+  expect(report).toEqual({ ok: true, issues: [] });
+});
+
+test("shared executable contract drives PayPal through its HTTP adapter", async () => {
+  globalThis.fetch = (async (value, init) => {
+    const url = String(value),
+      method = init?.method ?? "GET";
+    if (url.endsWith("/oauth2/token")) return json({ access_token: "access" });
+    if (url.endsWith("/checkout/orders") && method === "POST") {
+      const body = String(init?.body);
+      return json({
+        id: body.includes("cancel") ? "PP-CANCEL" : "PP-ORDER",
+        status: "CREATED",
+        links: [{ rel: "approve", href: "https://paypal.test/approve" }],
+      });
+    }
+    if (url.includes("/checkout/orders/"))
+      return json({
+        id: url.endsWith("PP-CANCEL") ? "PP-CANCEL" : "PP-ORDER",
+        status: "CREATED",
+        purchase_units: [{ amount: { value: "10.00", currency_code: "USD" } }],
+      });
+    if (url.includes("/refund")) return json({ id: "PP-REFUND", status: "COMPLETED" });
+    if (url.endsWith("verify-webhook-signature")) return json({ verification_status: "SUCCESS" });
+    return json({}, 404);
+  }) as typeof fetch;
+  const adapter = paypalGateway({
+    clientId: "client",
+    clientSecret: "paypal_secret",
+    webhookId: "hook",
+  });
+  const report = await exerciseGatewayContract(adapter, {
+    input,
+    webhook(intent: GatewayIntent) {
+      return new Request("https://test", {
+        method: "POST",
+        headers: {
+          "paypal-auth-algo": "SHA256withRSA",
+          "paypal-cert-url": "https://paypal.test/cert",
+          "paypal-transmission-id": "tx",
+          "paypal-transmission-sig": "sig",
+          "paypal-transmission-time": "time",
+        },
+        body: JSON.stringify({
+          id: "WH-1",
+          create_time: "2026-08-23T00:00:00Z",
+          event_type: "PAYMENT.CAPTURE.COMPLETED",
+          resource: {
+            id: "CAPTURE",
+            status: "COMPLETED",
+            supplementary_data: { related_ids: { order_id: intent.ref } },
+          },
+        }),
+      });
+    },
+  });
+  expect(report).toEqual({ ok: true, issues: [] });
+});

packages/payment/test/database-plugin.test.ts

@@ -20,7 +20,10 @@ test("payment plugin contributes its schema only when configured", async () => {
   const runner = createPluginRunner(paymentPlugin(), context);
   await runner.configure({ payment: { default: "sandbox", sandbox: true } } as any);
   const contributions = await runner.contributions();
-  expect(contributions.migrations.map((row) => row.id)).toEqual(["wrnexus-payment-001"]);
+  expect(contributions.migrations.map((row) => row.id)).toEqual([
+    "wrnexus-payment-001",
+    "wrnexus-payment-002-event-order",
+  ]);
   expect(contributions.componentDirs).toHaveLength(1);
   expect(contributions.routes.map((row) => row.path)).toEqual(["/api/payments/webhook/:gateway"]);
   const off = createPluginRunner(paymentPlugin(), { ...context, metadata: new Map() });
@@ -41,13 +44,14 @@ test("database store persists append-only events and atomically caps refunds", a
     warn() {},
   });
   await runner.configure({ payment: { default: "sandbox" } } as any);
-  const source = (await runner.contributions()).migrations[0].source!;
-  const up = source.split("-- +down")[0].replace("-- +up", "");
-  for (const statement of up
-    .split(";")
-    .map((x) => x.trim())
-    .filter(Boolean))
-    await db.exec(statement);
+  for (const migration of (await runner.contributions()).migrations) {
+    const up = migration.source!.split("-- +down")[0].replace("-- +up", "");
+    for (const statement of up
+      .split(";")
+      .map((x) => x.trim())
+      .filter(Boolean))
+      await db.exec(statement);
+  }
   const store = databasePaymentStore();
   const intent = {
     id: "pay-1",

packages/payment/test/payment.test.ts

@@ -8,9 +8,35 @@ import {
   sandboxGateway,
   signSandboxWebhook,
   stripeGateway,
+  derivePaymentStatus,
   verifyGatewayContract,
 } from "../src/index.ts";
 
+test("gateway occurrence order and sticky terminal states prevent webhook regression", () => {
+  const events = [
+    {
+      id: "refund",
+      intentId: "p",
+      type: "refunded",
+      status: "refunded" as const,
+      at: 100,
+      occurredAt: 300,
+    },
+    {
+      id: "late-capture-retry",
+      intentId: "p",
+      type: "captured",
+      status: "captured" as const,
+      at: 400,
+      occurredAt: 200,
+    },
+  ];
+  expect(derivePaymentStatus(events)).toBe("refunded");
+  expect(
+    derivePaymentStatus([...events, { ...events[1], id: "new-id", at: 500, occurredAt: 500 }]),
+  ).toBe("refunded");
+});
+
 test("sandbox drives webhook-authoritative capture and bounded partial refunds", async () => {
   const store = memoryPaymentStore(),
     gateway = sandboxGateway({ secret: "test-secret" });
@@ -69,6 +95,34 @@ test("sandbox drives webhook-authoritative capture and bounded partial refunds",
     idempotencyKey: "refund-2",
   });
   expect(excessive).toMatchObject({ ok: false, fault: "refused" });
+  expect(
+    (
+      await payment.refundPayment({
+        id: created.value.id,
+        amount: 600,
+        reason: "remaining return",
+        idempotencyKey: "refund-remaining",
+      })
+    ).ok,
+  ).toBe(true);
+  const stale = await signSandboxWebhook(
+    {
+      id: "evt-delayed-capture",
+      intentRef: created.value.gatewayRef,
+      type: "payment.captured",
+      status: "captured",
+      occurredAt: 1,
+    },
+    "test-secret",
+  );
+  await handler({
+    req: new Request("https://app.test/webhook", {
+      method: "POST",
+      headers: { "x-wrnexus-signature": stale.signature },
+      body: stale.body,
+    }),
+  });
+  expect(await payment.checkPayment(created.value.id)).toBe("refunded");
 });
 
 test("initialize, webhook, and refund replay change records once", async () => {