feat(payments): add deferred PostgreSQL payment state backend

.changeset/atomic-payment-state.md

@@ -0,0 +1,13 @@
+---
+'@contextvm/sdk': minor
+---
+
+Prevent duplicate payment offers and stale authorization changes using atomic, generation-fenced payment state and a retained transparent-request replay ledger.
+
+**Breaking change:** `AuthorizationStore` now implements the asynchronous `PaymentStateStore` contract (`admit`, `transition`, and `read`). Replace the former synchronous grant/claim/pending/clear methods and `maxEntries` option with the atomic store operations and `PaymentStatePolicy` limits. Inject the store through `ServerPaymentsOptions.stateStore`; both payment modes must use the same store.
+
+The supplied store is in memory: one paying server identity must run in one process with a shared store. Restarts lose its invoice bindings, grants, and replay records. PostgreSQL support is deferred to a separate backend branch.
+
+Server quotes and processor output are validated before payment or execution. Expired paid grants return `payment_expired` with instructions to reconcile the payment rather than request another payment automatically.
+
+Client and server boundaries share a non-empty, 2,048-character PMI schema. Identifiers are opaque and matched exactly, preserving custom rails and supporting HTTPS method identifiers without adding a protocol grammar restriction.

.github/workflows/test.yml

@@ -10,6 +10,19 @@ on:
 jobs:
   test:
     runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:17.7-bookworm
+        env:
+          POSTGRES_PASSWORD: contextvm-test
+          POSTGRES_DB: contextvm_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres -d contextvm_test"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
 
     steps:
       - uses: actions/checkout@v4
@@ -25,4 +38,5 @@ jobs:
       - name: Run tests
         env:
           NODE_ENV: test
+          PAYMENT_TEST_DATABASE_URL: postgres://postgres:contextvm-test@127.0.0.1:5432/contextvm_test
         run: bun test --concurrent --timeout=60000

package.json

@@ -72,6 +72,10 @@
     "./payments/*": {
       "types": "./dist/esm/payments/*.d.ts",
       "default": "./dist/esm/payments/*.js"
+    },
+    "./payments/postgres": {
+      "types": "./dist/esm/payments/postgres/index.d.ts",
+      "default": "./dist/esm/payments/postgres/index.js"
     }
   },
   "types": "./dist/esm/index.d.ts",
@@ -92,9 +96,11 @@
     "@changesets/cli": "^2.31.1",
     "@eslint/js": "^9.39.5",
     "@types/bun": "^1.4.2",
+    "@types/pg": "^8.23.1",
     "eslint": "^9.39.5",
     "eslint-config-prettier": "^10.1.8",
     "eslint-plugin-import": "^2.32.0",
+    "pg": "^8.23.1",
     "prettier": "^3.9.6",
     "typescript": "^5.9.3",
     "typescript-eslint": "^8.70.0"
@@ -109,5 +115,13 @@
     "rxjs": "^7.8.2",
     "ws": "^8.21.3",
     "zod": "^4.5.4"
+  },
+  "peerDependencies": {
+    "pg": "^8.23.1"
+  },
+  "peerDependenciesMeta": {
+    "pg": {
+      "optional": true
+    }
   }
 }

src/__mocks__/client-payments-harness.ts

@@ -0,0 +1,143 @@
+import type { Transport } from '@contextvm/mcp-sdk/shared/transport';
+import type {
+  JSONRPCMessage,
+  JSONRPCRequest,
+} from '@contextvm/mcp-sdk/types.js';
+import {
+  withClientPayments,
+  type ClientPaymentsOptions,
+} from '../payments/client-payments.js';
+import type { PaymentOption } from '../payments/types.js';
+import { waitFor as waitForValue } from '../core/utils/test.utils.js';
+
+/** Coordinates isolated test work without changing the global clock. */
+export async function waitFor(
+  condition: () => boolean,
+  options: { timeoutMs?: number } = {},
+): Promise<void> {
+  await waitForValue({
+    produce: () => (condition() ? true : undefined),
+    timeoutMs: options.timeoutMs ?? 1000,
+    intervalMs: 2,
+  });
+}
+
+/** Builds or advances isolated payment test state. */
+export const tick = (): Promise<void> =>
+  new Promise((resolve) => setTimeout(resolve, 0));
+/** Builds or advances isolated payment test state. */
+export const wait = (ms: number): Promise<void> =>
+  new Promise((resolve) => setTimeout(resolve, ms));
+/** Builds or advances isolated payment test state. */
+export const request = (id: string | number = 1): JSONRPCRequest => ({
+  jsonrpc: '2.0',
+  id,
+  method: 'tools/call',
+  params: { name: 'paid', arguments: { value: id } },
+});
+/** Builds or advances isolated payment test state. */
+export const option = (
+  pay_req = 'invoice',
+  amount = 5,
+  pmi = 'fake',
+): PaymentOption => ({ amount, pmi, pay_req });
+/** Builds or advances isolated payment test state. */
+export const offer = (input: unknown = option()): JSONRPCMessage =>
+  ({
+    jsonrpc: '2.0',
+    method: 'notifications/payment_required',
+    params: input,
+  }) as JSONRPCMessage;
+/** Builds or advances isolated payment test state. */
+export const required = (
+  id: string | number = 1,
+  options: unknown = [option()],
+): JSONRPCMessage => ({
+  jsonrpc: '2.0',
+  id,
+  error: {
+    code: -32042,
+    message: 'Payment Required',
+    data: { payment_options: options },
+  },
+});
+/** Builds or advances isolated payment test state. */
+export const pending = (
+  id: string | number = 1,
+  retry_after?: unknown,
+): JSONRPCMessage => ({
+  jsonrpc: '2.0',
+  id,
+  error: {
+    code: -32043,
+    message: 'Payment Pending',
+    data: retry_after === undefined ? {} : { retry_after },
+  },
+});
+/** Lets a test decide when an external payment action completes. */
+export function deferred<T>(): {
+  promise: Promise<T>;
+  resolve: (value: T) => void;
+  reject: (error: Error) => void;
+} {
+  let resolve!: (value: T) => void;
+  let reject!: (error: Error) => void;
+  const promise = new Promise<T>((yes, no) => {
+    resolve = yes;
+    reject = no;
+  });
+  return { promise, resolve, reject };
+}
+/** One transport, its wrapper, and all observable test outcomes. */
+export interface Harness {
+  paid: Transport;
+  base: Transport;
+  sent: JSONRPCMessage[];
+  received: JSONRPCMessage[];
+  errors: Error[];
+  deliver: (message: JSONRPCMessage) => void;
+}
+/** Coordinates isolated test work without changing the global clock. */
+export async function run(
+  options: ClientPaymentsOptions,
+  body: (harness: Harness) => Promise<void>,
+): Promise<void> {
+  const sent: JSONRPCMessage[] = [];
+  const received: JSONRPCMessage[] = [];
+  const errors: Error[] = [];
+  const base: Transport = {
+    async start(): Promise<void> {},
+    async send(message): Promise<void> {
+      sent.push(message);
+    },
+    async close(): Promise<void> {}, // Deliberately does not fire onclose.
+  };
+  const paid = withClientPayments(base, { minRetryDelayMs: 2, ...options });
+  paid.onmessage = (message): void => {
+    received.push(message);
+  };
+  paid.onerror = (error): void => {
+    errors.push(error);
+  };
+  await paid.start();
+  try {
+    await body({
+      paid,
+      base,
+      sent,
+      received,
+      errors,
+      deliver: (message): void => base.onmessage?.(message),
+    });
+  } finally {
+    await paid.close();
+  }
+}
+/** Collects terminal errors observed by the test caller. */
+export function failures(
+  harness: Harness,
+): Array<{ error: { data?: unknown } }> {
+  return harness.received.filter((message) => 'error' in message) as Array<{
+    error: { data?: unknown };
+  }>;
+}

src/__mocks__/payment-state-worker.ts

@@ -0,0 +1,49 @@
+import { Pool } from 'pg';
+import { PostgresPaymentStateStore } from '../payments/postgres/index.js';
+import {
+  paymentAuthorizationKey,
+  paymentEventKey,
+} from '../payments/payment-state-store.js';
+
+const [action, namespace] = process.argv.slice(2);
+const pool = new Pool({
+  connectionString: process.env.PAYMENT_TEST_DATABASE_URL,
+  max: 1,
+});
+const input = {
+  key:
+    action === 'claim'
+      ? paymentAuthorizationKey('client', 'hash')
+      : paymentEventKey('client', 'event-1'),
+  mode:
+    action === 'claim'
+      ? ('explicit_gating' as const)
+      : ('transparent' as const),
+  clientPubkey: 'client',
+  invocationHash: 'hash',
+  requestEventId: action === 'claim' ? 'worker-retry' : 'event-1',
+  leaseMs: 10000,
+};
+const store = new PostgresPaymentStateStore({ pool, namespace });
+await store.initialize();
+if (action === 'claim') {
+  console.log(JSON.stringify(await store.admit(input)));
+  await pool.end();
+} else if (action === 'invoice') {
+  const result = await store.admit(input);
+  if (result.status !== 'reserved')
+    throw new Error('Worker admission was refused');
+  const record = result.record;
+  const fence = { key: record.key, generation: record.generation };
+  await store.transition({ ...fence, type: 'issuing' });
+  await store.transition({
+    ...fence,
+    type: 'bind',
+    invoice: { pmi: 'fake', amount: 10, pay_req: 'persisted-worker-invoice' },
+    leaseMs: 10000,
+  });
+  console.log(JSON.stringify(await store.read(record.key)));
+  setInterval(() => {}, 1000);
+} else {
+  throw new Error('Unknown payment test worker action');
+}