Knex Storage

Persistence is provider-agnostic and pluggable through the StorageDriver contract. The shipped implementation is KnexStorageDriver, which works against any SQL database Knex supports.

Transaction-scoped audit events

KnexAuditLogRepository is exported for host applications that need to include their own domain mutation and an immutable audit entry in the same Knex transaction:

await knex.transaction(async (transaction) => {
  const repository = new KnexAuditLogRepository(transaction, clock, auditKey);
  await new AuditResource(repository, tenantId).record(event);
});

Use the same clock and audit key as KnexStorageDriver. Passing the transaction rather than the root Knex client ensures a thrown error rolls back both the host mutation and audit append. See Custom domain audit for a complete example.

The StorageDriver contract

Source: src/domain/contracts/storage-driver.contract.ts.

export interface Repositories {
  readonly canonicalProducts?: CanonicalProductRepository;
  readonly canonicalPrices?: CanonicalPriceRepository;
  readonly productProviderBindings?: ProductProviderBindingRepository;
  readonly priceProviderBindings?: PriceProviderBindingRepository;
  readonly customers: CustomerRepository;
  readonly products: ProductRepository;
  readonly prices: PriceRepository;
  readonly subscriptions: SubscriptionRepository;
  readonly subscriptionItems: SubscriptionItemRepository;
  readonly invoices: InvoiceRepository;
  readonly payments: PaymentRepository;
  readonly refunds: RefundRepository;
  readonly webhookEvents: WebhookEventRepository;
  readonly auditLogs: AuditLogRepository;
  readonly outboxEvents: OutboxEventRepository;
}

export interface StorageDriver extends Repositories {
  transaction<T>(work: (repos: Repositories) => Promise<T>): Promise<T>;
}

A StorageDriver exposes aggregate repositories plus a transaction method that runs a unit of work with a transactional copy of those same repositories. Canonical catalog repositories are optional for third-party driver compatibility; products() and prices() return a storage-required error when a driver does not implement them.

KnexStorageDriver

Source: src/infrastructure/storage/knex/knex-storage-driver.ts.

new KnexStorageDriver(knex: Knex, clock?: Clock, encryption?: Encryption);

• knex - a configured Knex instance.
• clock - defaults to SystemClock. Supplies the created_at / updated_at timestamps.
• encryption - optional; passed to the webhook-event repository for payload protection.

The constructor builds all repositories against the base connection. transaction(work) calls knex.transaction and rebuilds the repository set bound to the transaction handle trx, so every write inside the callback participates in the same transaction.

import knex from 'knex';
import { KnexStorageDriver } from '@akira-io/payable';

const db = knex({ client: 'pg', connection: process.env.DATABASE_URL });
const storage = new KnexStorageDriver(db);

await storage.transaction(async (repos) => {
  const customer = await repos.customers.create(/* ... */);
  await repos.subscriptions.create(/* ... */);
  return customer;
});

Catalog mutation durability

With KnexStorageDriver, the product, price, audit, and outbox repositories use the same Knex transaction. Entity, audit, and outbox writes occur atomically only when normalized durable state changes. Each changed mutation writes either its product or price row together with one audit record and one outbox event. A failure inside the transaction rolls back all three local writes. Identical state is a no-op with no update, audit record, or outbox event. The provider call occurs outside this SQL transaction and completes before the local writes begin.

The built-in product and price repositories use an internal compare-and-set operation for concurrent updates. After a transaction failure, Payable performs one read-after-failure by tenant, provider, and provider resource ID. If the normalized target is already durable, the call succeeds. Otherwise, Payable returns CATALOG_PERSISTENCE_FAILED with the remote identifier and correlation ID.

The internal compare-and-set method is not part of the public repository contracts. A third-party StorageDriver remains source-compatible and uses its normal transactional update method. It can participate in read-after-failure recovery through the existing findByProviderId contract, but it does not gain the built-in repositories’ conditional-update protection.

Schema

Tables are split into two groups, each created by a dedicated migration module. Every table name is prefixed payable_, every primary key is a uuid, and every table carries created_at / updated_at timestamps (except payable_audit_logs, which is append-only and has only created_at).

Billing schema

Source: src/infrastructure/storage/knex/migrations/billing-schema.ts. These hold the core billing aggregates.

Canonical local catalog schema

Source: src/infrastructure/storage/knex/migrations/canonical-catalog-schema.ts. Migration step 011-canonical-local-catalog creates these tables without changing legacy provider-first rows.

Local CRUD writes only canonical tables. Provider synchronization creates or updates bindings in a separate operation.

Migration step 017-canonical-provider-catalog-backfill upgrades beta6 provider-first catalog rows. It preserves legacy local IDs and accepted fields in the canonical tables, then creates separate bindings for non-null provider identifiers. The step rejects orphaned, cross-tenant, cross-provider, or conflicting rows before recording the ledger entry. It can resume after an interrupted batch without duplicating canonical rows or bindings. See Upgrading from 1.0.0-beta6 for verification and recovery.

Canonical subscription price migration schema

Migration step 021-canonical-subscription-price-migrations creates payable_subscription_price_migrations and payable_subscription_mutation_claims. The migration table stores tenant-qualified canonical subscription, source price, target price, and provider-binding references; immutable JSON snapshots; explicit timing and policies; lifecycle state; ownership; stable failures; and timestamps. Provider execution evidence is an internal persistence field and is never returned by the public resource or adapters. The claim table uses a stable claim reference, a database-enforced unique active tenant/subscription key, an exact owner token, a provider-neutral operation, opaque intent, and replayable observation and resolution fields. Step 022-subscription-mutation-recovery upgrades pre-recovery step-021 tables idempotently, including the opaque-intent rename and neutral observation columns.

Custom storage drivers must implement the complete public SubscriptionPriceMigrationRepository and SubscriptionMutationClaimRepository contracts. Execution evidence crosses the repository boundary only as the opaque SubscriptionPriceMigrationExecutionEvidenceBlob; provider fields and codecs remain internal. Persistent drivers use the public version-validating rehydration factory. Migration creation, tenant-scoped evidence reads, active lookup, reconciliation/settlement CAS, unique claim acquisition, tenant-scoped active lookup, exact-owner release, observation, reference lookup, and claim resolution are all required. This is an explicit custom-driver contract addition with no runtime cast or optional fallback.

The table enforces one active migration per tenant and subscription, exact scheduled-date shape, non-negative attempts, and execution-token correspondence. Indexes cover tenant/status pages, subscription pages, and due work ordered by effective_at, id. applied and cancelled release the active-subscription key. reconciliation_required retains it until explicit resolution, and pending_renewal retains it until explicit boundary settlement. Operator-confirmed not_applied resolves to failed while releasing the active key and shared mutation claim; a later retry must reacquire both fences.

Referential integrity

The split between hard foreign keys and plain indexed columns is deliberate, not an oversight:

• Composition relationships use real foreign keys with ON DELETE CASCADE. A payable_subscription_item cannot exist without its payable_subscription, and a payable_refund cannot exist without its payable_payment. Both children are created in-process alongside their parent, so the parent is always present and a cascade is the correct lifecycle.
• Cross-aggregate references are indexed columns with application-managed integrity. prices.product_id, subscriptions.customer_id, invoices.customer_id / subscription_id, and payments.customer_id point at aggregates that are populated by provider ingestion, which can arrive out of order (a subscription webhook may land before its customer is synced). A database foreign key would reject those inserts, so the application owns the integrity of these edges instead.

When deleting an aggregate root, the application is responsible for cleaning up the indexed references that do not cascade.

System schema

Source: src/infrastructure/storage/knex/migrations/system-schema.ts. These support webhooks, idempotency, audit, and the outbox.

migrate(knex)

Source: src/infrastructure/storage/knex/migrations/migrate.ts.

export async function migrate(knex: Knex): Promise<void> {
  await withMigrationLock(knex, async () => {
    await runStep(knex, '001-billing-tables', () => createBillingTables(knex));
    await runStep(knex, '002-system-tables', () => createSystemTables(knex));
    await runStep(knex, '003-alter-existing-tables', () => alterExistingTables(knex));
    await runStep(knex, '004-widen-endpoint-secret', () => widenEndpointSecret(knex));
    await runStep(knex, '005-webhook-occurred-at', () => addWebhookOccurredAt(knex));
    await runStep(knex, '006-subscription-provider-synced-at', () =>
      addSubscriptionProviderSyncedAt(knex),
    );
    await runStep(knex, '007-post-ledger-schema-convergence', () =>
      convergePostLedgerSchema(knex),
    );
    await runStep(knex, '008-customer-provider-bindings', () =>
      addCustomerProviderBindings(knex),
    );
    await runStep(knex, '009-catalog-tenant-keys', () => addCatalogTenantKeys(knex));
    await runStep(knex, '010-subscription-lifecycle-metadata', () =>
      addSubscriptionLifecycleMetadata(knex),
    );
    await runStep(knex, '011-canonical-local-catalog', () => addCanonicalCatalogTables(knex));
    await runStep(knex, '012-customer-provider-sync-states', () =>
      addCustomerProviderSyncStates(knex),
    );
    await runStep(knex, '013-customer-provider-sync-state-leases', () =>
      addCustomerProviderSyncStateLeases(knex),
    );
    await runStep(knex, '014-catalog-synchronization', () =>
      addCatalogSynchronizationTable(knex),
    );
    await runStep(knex, '015-canonical-local-subscriptions', () =>
      addCanonicalLocalSubscriptions(knex),
    );
    await runStep(knex, '016-provider-neutral-page-indexes', () =>
      addProviderNeutralPageIndexes(knex),
    );
    await runStep(knex, '017-canonical-provider-catalog-backfill', () =>
      backfillCanonicalProviderCatalog(knex),
    );
    await runStep(knex, '018-canonical-subscription-products', () =>
      addCanonicalSubscriptionProducts(knex).then(() => undefined),
    );
    await runStep(knex, '019-local-payment-evidence', () =>
      addLocalPaymentEvidence(knex),
    );
    await runStep(knex, '020-canonical-invoices', () =>
      addCanonicalInvoices(knex).then(() => undefined),
    );
    await runStep(knex, '021-canonical-subscription-price-migrations', () =>
      addCanonicalSubscriptionPriceMigrations(knex),
    );
    await runStep(knex, '022-subscription-mutation-recovery', () =>
      addSubscriptionMutationRecovery(knex),
    );
  });
}

All steps run inside withMigrationLock, which serializes concurrent migrators:

• On PostgreSQL it takes a session pg_advisory_lock and releases it in a finally.
• On MySQL / MariaDB it takes a named GET_LOCK (30s timeout); if the lock is not acquired it throws PayableError with code MIGRATION_LOCK_UNAVAILABLE.
• On any other dialect it runs the steps without a lock.

Each step is recorded through a migration ledger via runStep, so a completed step is skipped on a re-run. The first four steps establish the original schema; steps 005 through 007 add webhook timestamps, subscription sync timestamps, and post-ledger convergence. The customer identity migration is step 008. Step 011 adds the canonical catalog without rewriting legacy rows. Step 012 adds tenant-scoped customer provider synchronization lifecycle rows. Step 013 adds attempt_owner_id and lease_expires_at to existing synchronization state tables so attempts can be fenced and leased safely during retries. Later steps add catalog synchronization, canonical local subscriptions, provider-neutral page indexes, and the canonical provider catalog backfill. The complete sequence is:

1. Create billing tables (001-billing-tables) - each via createIfMissing (create-if-missing.ts), which checks knex.schema.hasTable(name) and only creates the table when it is absent.
2. Create system tables (002-system-tables) - same createIfMissing pattern.
3. Alter existing tables (003-alter-existing-tables) - alterExistingTables (alter-existing-tables.ts) performs additive migrations against already-created tables. ensureColumns adds a column only when knex.schema.hasColumn reports it missing (it back-fills normalized_type and data on payable_webhook_events for installations created before those columns existed). ensureIndexes issues CREATE INDEX IF NOT EXISTS for the composite keyset indexes: payable_subscriptions_customer_created_id_index, payable_invoices_customer_created_id_index, payable_payments_customer_created_id_index, payable_refunds_payment_created_id_index, and payable_outbox_events_status_locked_index.
4. Widen the endpoint secret (004-widen-endpoint-secret) - widenEndpointSecret (widen-endpoint-secret.ts) alters payable_webhook_endpoints.secret to text so the column can hold encrypted (sealed) secret values, which are longer than a raw secret.

• Customer provider bindings (008-customer-provider-bindings) - creates payable_customer_provider_bindings, backfills every non-null legacy provider customer id, verifies the backfill, and only then removes provider and provider_customer_id from payable_customers.
• Catalog tenant keys (009-catalog-tenant-keys) - migrates products and prices in this order: add non-null tenant_key with default ''; repeatedly select up to 100 mismatched rows ordered by id and backfill those ids until none remain; verify each value equals COALESCE(tenant_id, ''); add an enforced consistency check; reject duplicate (tenant_key, provider, provider id) rows with a non-null provider id; create the normalized unique index; then remove the legacy global index. The product index is payable_products_tenant_provider_product_unique. The price index is payable_prices_tenant_provider_price_unique.
• Canonical local catalog (011-canonical-local-catalog) - creates provider-neutral product and price tables plus their provider-binding tables. The step is additive and safe for databases whose earlier ledger entries are already applied.
• Customer provider sync leases (013-customer-provider-sync-state-leases) - adds nullable attempt_owner_id and lease_expires_at columns to synchronization tables created by step 012, including databases whose migration ledger already records step 012 as complete.
• Catalog synchronization (014-catalog-synchronization) - creates payable_catalog_synchronizations with a normalized tenant key, one lifecycle row per (tenant, provider, resource type, resource id), canonical generation and idempotency metadata, retry/reconciliation state, and nullable attempt owner and lease expiration columns.
• Canonical local subscriptions (015-canonical-local-subscriptions) - allows subscriptions without a provider, adds immutable accepted-price snapshot columns, normalizes the tenant key, and creates payable_subscription_provider_bindings. Existing provider identities are backfilled into tenant-scoped bindings without changing local subscription IDs.
• Provider-neutral page indexes (016-provider-neutral-page-indexes) - backfills normalized payment tenant keys, verifies their consistency, and creates deterministic collection indexes.
• Canonical provider catalog backfill (017-canonical-provider-catalog-backfill) - copies provider-first products and prices into canonical resources with the same local IDs, then creates bindings for non-null provider identifiers. Conflicting fields, orphaned prices, tenant mismatches, provider mismatches, and binding collisions stop the step before its ledger entry is recorded.
• Canonical subscription products (018-canonical-subscription-products) - stores the immutable canonical product identity accepted by each subscription.
• Local payment evidence (019-local-payment-evidence) - adds provider-neutral payment evidence used by canonical projections.
• Canonical invoices (020-canonical-invoices) - adds tenant-qualified canonical invoice storage.
• Canonical subscription price migrations (021-canonical-subscription-price-migrations) - adds immutable migration previews, lifecycle ownership, due-page indexes, tenant-qualified relations, and active-subscription uniqueness.
• Subscription mutation recovery (022-subscription-mutation-recovery) - upgrades opaque direct intent and neutral observation storage for exact retained-owner recovery.

Step 009-catalog-tenant-keys is fail-closed. The mismatch-driven batches revisit rows inserted below an earlier batch boundary. The consistency check validates existing rows when it is added and rejects later writes that omit the correct tenant key, including writes from an older application instance. A mismatch, failed check, or duplicate normalized identity throws before the matching legacy index is removed. The migration ledger entry is written only after both tables finish. Re-running resumes safely because existing columns, checks, and normalized indexes are retained.

migrate is idempotent and safe to re-run: it creates nothing that exists and adds only missing columns/indexes. A second migrate resolves cleanly, and a table created before the additive columns gets them back-filled.

import knex from 'knex';
import { migrate } from '@akira-io/payable';

const db = knex({ client: 'pg', connection: process.env.DATABASE_URL });
await migrate(db);

Upgrading through steps 021 and 022

Back up the database, deploy the Payable code that contains both steps, and run migrate(db) before a worker or request can create canonical migrations. The step is additive: it preserves subscriptions, subscription items, canonical catalog rows, and legacy preview tokens. It does not synthesize canonical migrations for historical provider-native subscriptions.

Run migrate(db) again after the first successful upgrade. The second run must leave the migration ledger and schema unchanged. If either step fails, do not write its ledger row by hand. Correct the reported foreign-key, tenant-identity, or index conflict and rerun the same migration entry point.

For a fresh database, the same migrate(db) call applies steps 001 through 022. Fresh creation, upgrade, and replay must converge on the same table, foreign keys, unique constraints, and page indexes.

Repositories

There is one Knex repository per aggregate, under src/infrastructure/storage/knex/repositories/: canonical product, canonical price, both canonical provider-binding repositories, knex-customer, legacy knex-product, legacy knex-price, knex-subscription, knex-subscription-item, knex-invoice, knex-payment, knex-refund, knex-webhook-event, knex-audit-log, knex-outbox-event, knex-idempotency, and the composed canonical subscription-price-migration repository.

They share a base class KnexRepository<Entity, New> (knex-repository.ts) providing:

• create(data) - generates a UUID via crypto.randomUUID(), stamps created_at / updated_at from the clock, inserts with .returning('*'), and falls back to findByIdOrFail when the driver does not return the row.
• createMany(data) - single batch insert (no-op on an empty array).
• update(id, patch) - updates with a fresh updated_at.
• findById(id, tenantId) and the protected firstWhere and manyWhere query helpers. Product and price repositories require tenantId; use null for the tenantless partition.

Each concrete repository supplies the table name and the toEntity / toRow column mappers. Shared column converters live in mappers.ts (toDate, toNullableDate, fromDate, toJson, fromJson, stripUndefined, toBool). stripUndefined is what lets partial updates skip untouched columns.

Keyset pagination

List queries use cursor (keyset) pagination, not offsets. Source: manyWhere in knex-repository.ts and the ListOptions contract (src/domain/contracts/list-options.contract.ts).

export interface ListCursor {
  createdAt: Date;
  id: string;
}

export interface ListOptions {
  limit?: number;
  before?: ListCursor;
}

Behavior and guarantees:

• Ordering is always created_at DESC, id DESC - newest first, with id as a deterministic tiebreaker. Rows come back newest-first.
• Cursor semantics: before is the last row of the previous page. The query fetches rows strictly older than the cursor using a compound predicate: created_at < cursor.createdAt OR (created_at = cursor.createdAt AND id < cursor.id). The tie clause on id is what prevents skipping or duplicating rows that share the same created_at.
• Limit: limit caps page size; omit it to fetch all matching rows.

Both the happy path (paging backwards through a list) and the boundary case hold: paging one row at a time through four rows with identical created_at returns all four exactly once, no skips.

Supported database clients

KnexStorageDriver is client-agnostic; it relies only on standard Knex schema and query building plus .returning('*'). The dev/test client is better-sqlite3, configured with client: 'better-sqlite3' and shipped in devDependencies. knex is an optional peer (>=3). Any Knex-supported SQL client (for example PostgreSQL via pg) works; install the matching driver in your application. pg is not bundled as a dependency of this package - add it yourself when targeting PostgreSQL.