diff --git a/CHANGELOG.md b/CHANGELOG.md index 52f8e111..19d4b38f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- Added a durable leased Stripe reconciliation worker that consumes verified-event + triggers through server-owned Subscription-to-organization authority, hashes + lease secrets at rest, prevents concurrent/stale completion, retries with bounded + exponential backoff, records append-only attempt evidence, and dead-letters work + after a finite attempt budget without persisting arbitrary provider error text. - Bound verified subscription-mode `checkout.session.completed` Customer and Subscription identities to exactly one server-recorded successful local Checkout Session before creating reconciliation work; missing or ambiguous Session diff --git a/docs/doctoring/stripe-reconciliation-worker.md b/docs/doctoring/stripe-reconciliation-worker.md new file mode 100644 index 00000000..49d9a254 --- /dev/null +++ b/docs/doctoring/stripe-reconciliation-worker.md @@ -0,0 +1,70 @@ +# Stripe reconciliation worker: leased consumption, bounded retry, and dead-letter evidence + +## Status and shipped-truth boundary + +This record describes the **active PR** stacked on the current Checkout identity-bootstrap slice. Protected `develop` does not ship this worker until the complete prerequisite #488 stack is integrated under live branch protection and exact-head evidence. The parent stack already authenticates Stripe webhook bytes, persists immutable verified events, derives bounded reconciliation triggers, re-fetches current Subscription/Invoice authority, persists observations and entitlement claims, and binds a first Subscription identity from one successful local Checkout attempt. This slice adds only the durable consumption boundary for those queued triggers. + +A verified webhook is still not lifecycle or entitlement authority. Stripe explicitly documents automatic retries and non-guaranteed event ordering, and recommends retrieving missing/current objects instead of depending on webhook arrival order. The worker therefore leases a trigger and calls the existing authoritative reconciliation service, which re-fetches current Stripe state before any claim decision is persisted. + +## Decision + +Use two normalized worker relations in addition to the immutable `billing_stripe_reconciliation_triggers` relation: + +- `billing_stripe_reconciliation_jobs` is one mutable scheduling head per verified event trigger. It records `pending`, `processing`, `succeeded`, or `dead_letter`, bounded attempt count, next eligible time, a SHA-256 lease-token digest, lease expiry, completion time, a stable machine error code, and the final claim decision identity when successful. +- `billing_stripe_reconciliation_attempts` is append-only attempt evidence. It records the attempt number, lease window, terminal outcome, and bounded machine error code without persisting the plaintext lease secret or arbitrary provider exception text. + +The existing trigger table remains immutable event-to-Subscription work identity. Worker jobs are lazily seeded from triggers during claim so triggers that predate worker deployment are not stranded. + +## Authority chain + +The worker does not accept an organization identifier from a caller. After claiming a Subscription trigger, it resolves organization authority through the existing normalized `billing_stripe_subscriptions` → `billing_stripe_customers.organization_id` relation. If that authority is not present yet, the job remains explicit retryable work with `stripe_reconciliation_authority_missing`; the provider reconciliation port is not called. + +For a resolved tenant, `runNextStripeReconciliationJob` invokes the existing `reconcileStripeBillingAuthoritatively` boundary with exactly the server-derived organization, claimed Subscription, and verified event ID as provenance. The returned receipt is revalidated against that tenant and Subscription before the lease can complete. The worker never writes `orgs.plan`, memberships, RBAC, browser sessions, or capabilities. + +## Lease and concurrency contract + +Each claim creates a finite lease with an opaque random token. Only the SHA-256 digest is durable. The plaintext token exists only in the claiming process and is required for compare-and-set completion or failure. + +An unexpired lease excludes another worker. At or after exact lease expiry, a later claim records the abandoned attempt as `retry` and makes the job eligible again. A stale worker cannot complete or fail a reclaimed lease. If a lease expires on the final configured attempt, the job is moved to `dead_letter` instead of being left permanently `pending` but unclaimable. + +Claim, lease-expiry repair, completion, retry, dead-letter transition, and corresponding attempt evidence each execute inside a named SQLite savepoint. SQLite documents that savepoints may be nested and that `ROLLBACK TO` rewinds to the savepoint while leaving it active until `RELEASE`; the implementation therefore releases after success, and after failure releases only when rollback was confirmed. Cleanup failure after confirmed rollback never replaces the causal failure. + +## Retry and operational contract + +Provider, persistence, and reconciliation failures are bounded by a finite attempt budget. Retry delay is exponential with an explicit ceiling, preventing a hot failure loop. Only stable machine-readable `stripe_*` error codes are retained from downstream failures; arbitrary exception messages are collapsed to `stripe_reconciliation_failed`, preventing provider response text or secret-like values from entering durable worker evidence. + +The final-attempt state is `dead_letter`, not silent drop. This is deliberate operator-visible recovery evidence. A subsequent slice may add an authenticated operator inspection/requeue surface, but this worker does not create such authority implicitly. + +## TDD and acceptance traceability + +`tests/unit/stripe-reconciliation-worker.test.mjs` began as a test-only commit importing the absent production module, creating a deterministic module-resolution RED before implementation. Current behavior exercises real in-memory SQLite relations and requires: + +1. exactly one trigger claim, server-owned tenant resolution, exact event provenance into authoritative reconciliation, and durable success/attempt evidence; +2. exclusion under an unexpired lease, reclaim after expiry, and rejection of stale first-worker completion; +3. capped retry with a finite dead-letter budget and no persistence of arbitrary provider/secret-like exception text; +4. explicit retry when Subscription tenant identity is not yet available, without invoking provider reconciliation; and +5. use of the actual production Stripe customer/subscription schema rather than a test-only alias, preventing schema-drift false greens. + +`package.json` places the worker regression in normal unit CI and canonical c8 cases. `tests/unit/coverage-script-contract.test.mjs` locks both the production module instrumentation and the behavior test registration against silent removal. `server/db.mjs` installs the worker schema only after its trigger/evidence prerequisites and exports a configured `reconcileNextStripeBillingTrigger()` bootstrap boundary. + +## Privacy, security, and audit implications + +The worker stores provider event and Subscription identifiers already present in the billing evidence model, scheduling timestamps, bounded machine error codes, claim decision IDs, and lease-token hashes. It stores no raw Stripe payload, webhook body, Stripe secret, session token, plaintext worker lease token, email address, or payment method data. The design supports purpose-bound operational evidence and least-privilege processing without claiming SOC 2, CSAP, PCI DSS, or other certification. + +## Rollback + +Rollback removes `server/stripe_reconciliation_worker.mjs`, its bootstrap wiring, tests/coverage registration, this doctoring record, and its Unreleased changelog entry together. Before protected integration there is no production data migration. After integration, rollback must preserve the immutable verified events/triggers and worker job/attempt tables as evidence unless an approved migration explicitly proves safe archival; deleting failed/retry evidence is not a rollback strategy. + +## Remaining #488 work + +This slice intentionally does not run a perpetual scheduler, expose dead-letter recovery UI/API, define long-term retention/export policy, or complete final out-of-order end-to-end acceptance. Those remain subsequent bounded integration/recovery/release slices. External Stripe delivery latency never becomes ordering authority; successful reconciliation still depends on current provider reads and existing monotonic claim logic. + +## References + +Stripe. (2026). *Receive Stripe events in your webhook endpoint*. Stripe Documentation. https://docs.stripe.com/webhooks + +Stripe. (2026). *Process undelivered webhook events*. Stripe Documentation. https://docs.stripe.com/webhooks/process-undelivered-events + +SQLite Consortium. (2026). *Savepoints*. SQLite Documentation. https://www.sqlite.org/lang_savepoint.html + +SQLite Consortium. (2026). *Transaction*. SQLite Documentation. https://www.sqlite.org/lang_transaction.html diff --git a/package.json b/package.json index 0deb26ee..15937831 100644 --- a/package.json +++ b/package.json @@ -13,9 +13,9 @@ "coverage": "npm run test:coverage", "server": "node server/server.mjs", "test:api": "node tests/api/auth-secret.test.mjs && node --env-file=tests/api/smoke.env tests/api/smoke.mjs && node tests/api/ratelimit.test.mjs && node tests/api/attachment-status.test.mjs && node tests/api/session-revocation.test.mjs && node tests/api/orchestrator-attribution.test.mjs && node tests/api/billing-checkout.test.mjs && node tests/api/billing-live-checkout.test.mjs && node tests/api/billing-effective-plan-status.test.mjs && node tests/api/stripe-webhook.test.mjs", - "test:unit": "node tests/unit/opencode-config.test.mjs && node tests/unit/changelog-release-notes.test.mjs && node tests/unit/analytics.test.mjs && node tests/unit/cpm.test.mjs && node tests/unit/baseline-compare.test.mjs && node tests/unit/workload.test.mjs && node tests/unit/cost-evm.test.mjs && node tests/unit/msproject.test.mjs && node tests/unit/auth-password.test.mjs && node tests/unit/editor-unsaved.test.mjs && node tests/unit/static-coverage-evidence.test.mjs && node tests/unit/dep-types.test.mjs && node tests/unit/weekly-report.test.mjs && node tests/unit/clearfolio.test.mjs && node tests/unit/clearfolio-adapter-mock-hmac.test.mjs && node tests/unit/orchestrator.test.mjs && node tests/unit/orchestrator-coverage.test.mjs && node tests/unit/orchestrator-attribution.test.mjs && node tests/unit/sprint-stats.test.mjs && node tests/unit/burndown.test.mjs && node tests/unit/pm-analysis.test.mjs && node tests/unit/cloud-sync-security.test.mjs && node tests/unit/attachment-status.test.mjs && node tests/unit/clearfolio-status-signal.test.mjs && node tests/unit/coverage-script-contract.test.mjs && node tests/unit/billing-configuration.test.mjs && node tests/unit/billing-effective-plan.test.mjs && node tests/unit/billing-effective-plan-bootstrap.test.mjs && node tests/unit/billing-status-package-contract.test.mjs && node tests/unit/billing-status-response.test.mjs && node tests/unit/billing-checkout-attempt.test.mjs && node tests/unit/billing-checkout-attempt-authority.test.mjs && node tests/unit/billing-checkout-reconciliation.test.mjs && node tests/unit/billing-checkout-reconciliation-authority.test.mjs && node tests/unit/billing-checkout.test.mjs && node tests/unit/billing-provider-boundary.test.mjs && node tests/unit/billing-checkout-review-regressions.test.mjs && node tests/unit/stripe-webhook-boundary.test.mjs && node tests/unit/stripe-webhook-event-ledger.test.mjs && node tests/unit/stripe-webhook-recorder-integration.test.mjs && node tests/unit/stripe-subscription-provider.test.mjs && node tests/unit/stripe-subscription-metadata-propagation.test.mjs && node tests/unit/stripe-subscription-observation-ledger.test.mjs && node tests/unit/stripe-subscription-current-projection.test.mjs && node tests/unit/stripe-entitlement-policy.test.mjs && node tests/unit/stripe-entitlement-policy-edge.test.mjs && node tests/unit/stripe-entitlement-policy-duplicate-claims.test.mjs && node tests/unit/stripe-invoice-provider.test.mjs && node tests/unit/stripe-invoice-provider-edge.test.mjs && node tests/unit/stripe-invoice-observation-ledger.test.mjs && node tests/unit/stripe-invoice-current-projection-package-contract.test.mjs && node tests/unit/stripe-invoice-current-projection.test.mjs && node tests/unit/stripe-entitlement-claim-package-contract.test.mjs && node tests/unit/stripe-entitlement-claim-ledger.test.mjs && node tests/unit/stripe-billing-authoritative-reconciliation.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap-package-contract.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue-integration.test.mjs && node tests/unit/toast-accessibility.test.mjs", - "test:coverage": "c8 --all --include=app.js --include=cloud-sync.js --include=scripts/ci/static_coverage_evidence.mjs --include=server/attachment_status.mjs --include=server/app.mjs --include=server/application_routes.mjs --include=server/auth.mjs --include=server/billing.mjs --include=server/billing_checkout_attempt.mjs --include=server/billing_configuration.mjs --include=server/billing_status_response.mjs --include=server/clearfolio.mjs --include=server/orchestrator.mjs --include=server/stripe_webhook.mjs --include=server/stripe_webhook_event_ledger.mjs --include=server/stripe_subscription_provider.mjs --include=server/stripe_subscription_observation_ledger.mjs --include=server/stripe_subscription_current_projection.mjs --include=server/stripe_entitlement_policy.mjs --include=server/stripe_invoice_provider.mjs --include=server/stripe_invoice_observation_ledger.mjs --include=server/stripe_invoice_current_projection.mjs --include=server/stripe_entitlement_claim_ledger.mjs --include=server/stripe_billing_reconciliation.mjs --include=server/stripe_webhook_reconciliation_queue.mjs --include=server/stripe_checkout_identity_bootstrap.mjs --reporter=json --reporter=json-summary npm run test:coverage:cases", - "test:coverage:cases": "node tests/unit/coverage-script-contract.test.mjs && node tests/unit/attachment-status.test.mjs && node tests/unit/clearfolio-status-signal.test.mjs && node tests/unit/clearfolio-adapter-mock-hmac.test.mjs && node tests/unit/orchestrator.test.mjs && node tests/unit/orchestrator-coverage.test.mjs && node tests/unit/orchestrator-attribution.test.mjs && node tests/unit/msproject.test.mjs && node tests/unit/auth-password.test.mjs && node tests/unit/editor-unsaved.test.mjs && node tests/unit/static-coverage-evidence.test.mjs && node tests/unit/billing-configuration.test.mjs && node tests/unit/billing-effective-plan.test.mjs && node tests/unit/billing-effective-plan-bootstrap.test.mjs && node tests/unit/billing-status-package-contract.test.mjs && node tests/unit/billing-status-response.test.mjs && node tests/unit/billing-checkout-attempt.test.mjs && node tests/unit/billing-checkout-attempt-authority.test.mjs && node tests/unit/billing-checkout-reconciliation.test.mjs && node tests/unit/billing-checkout-reconciliation-authority.test.mjs && node tests/unit/billing-checkout.test.mjs && node tests/unit/billing-provider-boundary.test.mjs && node tests/unit/billing-checkout-review-regressions.test.mjs && node tests/unit/stripe-webhook-boundary.test.mjs && node tests/unit/stripe-webhook-event-ledger.test.mjs && node tests/unit/stripe-webhook-recorder-integration.test.mjs && node tests/unit/stripe-subscription-provider.test.mjs && node tests/unit/stripe-subscription-metadata-propagation.test.mjs && node tests/unit/stripe-subscription-observation-ledger.test.mjs && node tests/unit/stripe-subscription-current-projection.test.mjs && node tests/unit/stripe-entitlement-policy.test.mjs && node tests/unit/stripe-entitlement-policy-edge.test.mjs && node tests/unit/stripe-entitlement-policy-duplicate-claims.test.mjs && node tests/unit/stripe-invoice-provider.test.mjs && node tests/unit/stripe-invoice-provider-edge.test.mjs && node tests/unit/stripe-invoice-observation-ledger.test.mjs && node tests/unit/stripe-invoice-current-projection-package-contract.test.mjs && node tests/unit/stripe-invoice-current-projection.test.mjs && node tests/unit/stripe-entitlement-claim-package-contract.test.mjs && node tests/unit/stripe-entitlement-claim-ledger.test.mjs && node tests/unit/stripe-billing-authoritative-reconciliation.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue-integration.test.mjs && npm run test:api", + "test:unit": "node tests/unit/opencode-config.test.mjs && node tests/unit/changelog-release-notes.test.mjs && node tests/unit/analytics.test.mjs && node tests/unit/cpm.test.mjs && node tests/unit/baseline-compare.test.mjs && node tests/unit/workload.test.mjs && node tests/unit/cost-evm.test.mjs && node tests/unit/msproject.test.mjs && node tests/unit/auth-password.test.mjs && node tests/unit/editor-unsaved.test.mjs && node tests/unit/static-coverage-evidence.test.mjs && node tests/unit/dep-types.test.mjs && node tests/unit/weekly-report.test.mjs && node tests/unit/clearfolio.test.mjs && node tests/unit/clearfolio-adapter-mock-hmac.test.mjs && node tests/unit/orchestrator.test.mjs && node tests/unit/orchestrator-coverage.test.mjs && node tests/unit/orchestrator-attribution.test.mjs && node tests/unit/sprint-stats.test.mjs && node tests/unit/burndown.test.mjs && node tests/unit/pm-analysis.test.mjs && node tests/unit/cloud-sync-security.test.mjs && node tests/unit/attachment-status.test.mjs && node tests/unit/clearfolio-status-signal.test.mjs && node tests/unit/coverage-script-contract.test.mjs && node tests/unit/billing-configuration.test.mjs && node tests/unit/billing-effective-plan.test.mjs && node tests/unit/billing-effective-plan-bootstrap.test.mjs && node tests/unit/billing-status-package-contract.test.mjs && node tests/unit/billing-status-response.test.mjs && node tests/unit/billing-checkout-attempt.test.mjs && node tests/unit/billing-checkout-attempt-authority.test.mjs && node tests/unit/billing-checkout-reconciliation.test.mjs && node tests/unit/billing-checkout-reconciliation-authority.test.mjs && node tests/unit/billing-checkout.test.mjs && node tests/unit/billing-provider-boundary.test.mjs && node tests/unit/billing-checkout-review-regressions.test.mjs && node tests/unit/stripe-webhook-boundary.test.mjs && node tests/unit/stripe-webhook-event-ledger.test.mjs && node tests/unit/stripe-webhook-recorder-integration.test.mjs && node tests/unit/stripe-subscription-provider.test.mjs && node tests/unit/stripe-subscription-metadata-propagation.test.mjs && node tests/unit/stripe-subscription-observation-ledger.test.mjs && node tests/unit/stripe-subscription-current-projection.test.mjs && node tests/unit/stripe-entitlement-policy.test.mjs && node tests/unit/stripe-entitlement-policy-edge.test.mjs && node tests/unit/stripe-entitlement-policy-duplicate-claims.test.mjs && node tests/unit/stripe-invoice-provider.test.mjs && node tests/unit/stripe-invoice-provider-edge.test.mjs && node tests/unit/stripe-invoice-observation-ledger.test.mjs && node tests/unit/stripe-invoice-current-projection-package-contract.test.mjs && node tests/unit/stripe-invoice-current-projection.test.mjs && node tests/unit/stripe-entitlement-claim-package-contract.test.mjs && node tests/unit/stripe-entitlement-claim-ledger.test.mjs && node tests/unit/stripe-billing-authoritative-reconciliation.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap-package-contract.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue-integration.test.mjs && node tests/unit/stripe-reconciliation-worker.test.mjs && node tests/unit/stripe-reconciliation-worker-time-budget.test.mjs && node tests/unit/toast-accessibility.test.mjs", + "test:coverage": "c8 --all --include=app.js --include=cloud-sync.js --include=scripts/ci/static_coverage_evidence.mjs --include=server/attachment_status.mjs --include=server/app.mjs --include=server/application_routes.mjs --include=server/auth.mjs --include=server/billing.mjs --include=server/billing_checkout_attempt.mjs --include=server/billing_configuration.mjs --include=server/billing_status_response.mjs --include=server/clearfolio.mjs --include=server/orchestrator.mjs --include=server/stripe_webhook.mjs --include=server/stripe_webhook_event_ledger.mjs --include=server/stripe_subscription_provider.mjs --include=server/stripe_subscription_observation_ledger.mjs --include=server/stripe_subscription_current_projection.mjs --include=server/stripe_entitlement_policy.mjs --include=server/stripe_invoice_provider.mjs --include=server/stripe_invoice_observation_ledger.mjs --include=server/stripe_invoice_current_projection.mjs --include=server/stripe_entitlement_claim_ledger.mjs --include=server/stripe_billing_reconciliation.mjs --include=server/stripe_webhook_reconciliation_queue.mjs --include=server/stripe_checkout_identity_bootstrap.mjs --include=server/stripe_reconciliation_worker.mjs --reporter=json --reporter=json-summary npm run test:coverage:cases", + "test:coverage:cases": "node tests/unit/coverage-script-contract.test.mjs && node tests/unit/attachment-status.test.mjs && node tests/unit/clearfolio-status-signal.test.mjs && node tests/unit/clearfolio-adapter-mock-hmac.test.mjs && node tests/unit/orchestrator.test.mjs && node tests/unit/orchestrator-coverage.test.mjs && node tests/unit/orchestrator-attribution.test.mjs && node tests/unit/msproject.test.mjs && node tests/unit/auth-password.test.mjs && node tests/unit/editor-unsaved.test.mjs && node tests/unit/static-coverage-evidence.test.mjs && node tests/unit/billing-configuration.test.mjs && node tests/unit/billing-effective-plan.test.mjs && node tests/unit/billing-effective-plan-bootstrap.test.mjs && node tests/unit/billing-status-package-contract.test.mjs && node tests/unit/billing-status-response.test.mjs && node tests/unit/billing-checkout-attempt.test.mjs && node tests/unit/billing-checkout-attempt-authority.test.mjs && node tests/unit/billing-checkout-reconciliation.test.mjs && node tests/unit/billing-checkout-reconciliation-authority.test.mjs && node tests/unit/billing-checkout.test.mjs && node tests/unit/billing-provider-boundary.test.mjs && node tests/unit/billing-checkout-review-regressions.test.mjs && node tests/unit/stripe-webhook-boundary.test.mjs && node tests/unit/stripe-webhook-event-ledger.test.mjs && node tests/unit/stripe-webhook-recorder-integration.test.mjs && node tests/unit/stripe-subscription-provider.test.mjs && node tests/unit/stripe-subscription-metadata-propagation.test.mjs && node tests/unit/stripe-subscription-observation-ledger.test.mjs && node tests/unit/stripe-subscription-current-projection.test.mjs && node tests/unit/stripe-entitlement-policy.test.mjs && node tests/unit/stripe-entitlement-policy-edge.test.mjs && node tests/unit/stripe-entitlement-policy-duplicate-claims.test.mjs && node tests/unit/stripe-invoice-provider.test.mjs && node tests/unit/stripe-invoice-provider-edge.test.mjs && node tests/unit/stripe-invoice-observation-ledger.test.mjs && node tests/unit/stripe-invoice-current-projection-package-contract.test.mjs && node tests/unit/stripe-invoice-current-projection.test.mjs && node tests/unit/stripe-entitlement-claim-package-contract.test.mjs && node tests/unit/stripe-entitlement-claim-ledger.test.mjs && node tests/unit/stripe-billing-authoritative-reconciliation.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue-integration.test.mjs && node tests/unit/stripe-reconciliation-worker.test.mjs && node tests/unit/stripe-reconciliation-worker-time-budget.test.mjs && npm run test:api", "test:e2e": "playwright test", "test:e2e:headed": "playwright test --headed", "test:e2e:cloud": "playwright install chromium && playwright test tests/e2e/cloud.spec.js tests/e2e/toast-accessibility.spec.js", diff --git a/server/db.mjs b/server/db.mjs index 7386762c..bf18db64 100644 --- a/server/db.mjs +++ b/server/db.mjs @@ -8,6 +8,12 @@ import { createSqliteBillingCheckoutAttemptRepository, installBillingCheckoutAttemptSchema, } from './billing_checkout_attempt.mjs'; +import { reconcileStripeBillingAuthoritatively } from './stripe_billing_reconciliation.mjs'; +import { + createSqliteStripeReconciliationWorkerRepository, + installStripeReconciliationWorkerSchema, + runNextStripeReconciliationJob, +} from './stripe_reconciliation_worker.mjs'; import { configureStripeWebhookEventRecorder, createSqliteStripeWebhookEventRepository, @@ -255,6 +261,29 @@ installStripeEntitlementClaimSchema(db); export const stripeEntitlementClaims = createSqliteStripeEntitlementClaimRepository(db, { deriveEntitlement: deriveStripeSubscriptionEntitlement, }); +installStripeReconciliationWorkerSchema(db); +export const stripeReconciliationWorker = createSqliteStripeReconciliationWorkerRepository(db); + +/** + * Consume at most one pending verified Stripe reconciliation trigger. + * + * This bootstrap wrapper keeps queue leasing and authoritative provider reads behind + * server-owned repositories. Callers cannot choose tenant authority or inject claim + * identities; an idle queue returns `{ status: 'idle' }`. + * + * @returns {Promise>} bounded worker result + */ +export function reconcileNextStripeBillingTrigger() { + return runNextStripeReconciliationJob({ + repository: stripeReconciliationWorker, + reconcile: reconcileStripeBillingAuthoritatively, + reconciliationDependencies: { + subscriptionRepository: stripeSubscriptionObservations, + invoiceRepository: stripeInvoiceObservations, + claimRepository: stripeEntitlementClaims, + }, + }); +} // node:sqlite returns lastInsertRowid as number|bigint; normalize to Number. export const rowid = (r) => Number(r.lastInsertRowid); diff --git a/server/stripe_reconciliation_worker.mjs b/server/stripe_reconciliation_worker.mjs new file mode 100644 index 00000000..3cef2672 --- /dev/null +++ b/server/stripe_reconciliation_worker.mjs @@ -0,0 +1,590 @@ +import { createHash, randomUUID } from 'node:crypto'; + +const MAX_PROVIDER_ID_LENGTH = 255; +const MAX_ERROR_CODE_LENGTH = 96; +const EVENT_ID_PATTERN = /^[A-Za-z0-9_:-]+$/u; +const SUBSCRIPTION_ID_PATTERN = /^sub_[A-Za-z0-9_]+$/u; +const ERROR_CODE_PATTERN = /^[a-z0-9_:-]+$/u; +const LEASE_TOKEN_PATTERN = /^[A-Za-z0-9_-]{16,128}$/u; +const SAVEPOINT_NAME = 'billing_stripe_reconciliation_worker_write'; +const LEASE_EXPIRED_CODE = 'stripe_reconciliation_lease_expired'; +const DEFAULT_LEASE_MS = 90_000; +const DEFAULT_MAX_ATTEMPTS = 5; +const DEFAULT_BASE_BACKOFF_MS = 5_000; +const DEFAULT_MAX_BACKOFF_MS = 300_000; + +/** Stable fail-closed error for durable Stripe reconciliation worker operations. */ +export class StripeReconciliationWorkerError extends Error { + /** + * Create one sanitized worker failure. + * @param {string} code stable machine-readable failure code + * @param {number} [status=400] HTTP-compatible status for a future adapter + */ + constructor(code, status = 400) { + super(code); + this.name = 'StripeReconciliationWorkerError'; + this.code = code; + this.status = status; + } +} + +function workerError(code, status = 400) { + return new StripeReconciliationWorkerError(code, status); +} + +function boundedIdentifier(value, pattern) { + if ( + typeof value !== 'string' + || value.length === 0 + || value.length > MAX_PROVIDER_ID_LENGTH + || !pattern.test(value) + ) { + throw workerError('stripe_reconciliation_worker_invalid'); + } + return value; +} + +function positiveInteger(value, name) { + if (!Number.isSafeInteger(value) || value <= 0) { + throw new TypeError(`${name} must be a positive safe integer`); + } + return value; +} + +function nonNegativeInteger(value, name) { + if (!Number.isSafeInteger(value) || value < 0) { + throw new TypeError(`${name} must be a non-negative safe integer`); + } + return value; +} + +function positiveOption(value, fallback, name) { + return value === undefined ? fallback : positiveInteger(value, name); +} + +function normalizedNow(now) { + const value = Number(now()); + if (!Number.isSafeInteger(value) || value < 0) { + throw workerError('stripe_reconciliation_worker_clock_invalid', 500); + } + return value; +} + +function leaseTokenValue(randomToken) { + const value = randomToken(); + if (typeof value !== 'string' || !LEASE_TOKEN_PATTERN.test(value)) { + throw workerError('stripe_reconciliation_worker_token_invalid', 500); + } + return value; +} + +function tokenHash(value) { + const token = boundedIdentifier(value, LEASE_TOKEN_PATTERN); + return createHash('sha256').update(token, 'utf8').digest('hex'); +} + +function boundedFailureCode(value) { + if ( + typeof value !== 'string' + || value.length === 0 + || value.length > MAX_ERROR_CODE_LENGTH + || !ERROR_CODE_PATTERN.test(value) + ) { + throw workerError('stripe_reconciliation_worker_invalid'); + } + return value; +} + +function safeFailureCode(error) { + const code = error && typeof error === 'object' ? error.code : null; + if ( + typeof code === 'string' + && code.startsWith('stripe_') + && code.length <= MAX_ERROR_CODE_LENGTH + && ERROR_CODE_PATTERN.test(code) + ) { + return code; + } + return 'stripe_reconciliation_failed'; +} + +function safeAdd(left, right) { + const sum = left + right; + if ( + !Number.isSafeInteger(left) + || !Number.isSafeInteger(right) + || left < 0 + || right < 0 + || !Number.isSafeInteger(sum) + ) { + throw workerError('stripe_reconciliation_worker_clock_invalid', 500); + } + return sum; +} + +function retryDelay(attemptNumber, baseBackoffMs, maxBackoffMs) { + const exponent = Math.min(attemptNumber - 1, 30); + const scaled = baseBackoffMs * (2 ** exponent); + return Number.isFinite(scaled) ? Math.min(scaled, maxBackoffMs) : maxBackoffMs; +} + +function withSavepoint(database, operation) { + database.exec(`SAVEPOINT ${SAVEPOINT_NAME}`); + try { + const result = operation(); + database.exec(`RELEASE SAVEPOINT ${SAVEPOINT_NAME}`); + return result; + } catch (error) { + let rolledBack = false; + try { + database.exec(`ROLLBACK TO SAVEPOINT ${SAVEPOINT_NAME}`); + rolledBack = true; + } catch { + // Leave an unconfirmed failed savepoint open instead of risking partial commit. + } + if (rolledBack) { + try { + database.exec(`RELEASE SAVEPOINT ${SAVEPOINT_NAME}`); + } catch { + // Cleanup after confirmed rollback must never replace the causal failure. + } + } + throw error; + } +} + +/** + * Install durable job-head and append-only attempt tables for Stripe reconciliation. + * + * The immutable webhook trigger stays the source of work identity. The worker tables + * contain only scheduling/audit metadata and a hash of the active lease secret; they + * never copy raw provider payloads, webhook bodies, API secrets, or session authority. + * + * @param {import('node:sqlite').DatabaseSync} database bootstrapped SQLite database + * @returns {void} + */ +export function installStripeReconciliationWorkerSchema(database) { + if (!database || typeof database.exec !== 'function') { + throw new TypeError('database must provide SQLite exec operations'); + } + database.exec(` + CREATE TABLE IF NOT EXISTS billing_stripe_reconciliation_jobs ( + event_id TEXT PRIMARY KEY + REFERENCES billing_stripe_reconciliation_triggers(event_id) ON DELETE CASCADE, + processing_state TEXT NOT NULL DEFAULT 'pending' + CHECK(processing_state IN ('pending','processing','succeeded','dead_letter')), + attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0), + next_attempt_at_ms INTEGER NOT NULL CHECK(next_attempt_at_ms >= 0), + lease_token_sha256 TEXT + CHECK(lease_token_sha256 IS NULL OR length(lease_token_sha256) = 64), + lease_expires_at_ms INTEGER CHECK(lease_expires_at_ms IS NULL OR lease_expires_at_ms >= 0), + completed_at_ms INTEGER CHECK(completed_at_ms IS NULL OR completed_at_ms >= 0), + last_error_code TEXT + CHECK(last_error_code IS NULL OR length(last_error_code) BETWEEN 1 AND ${MAX_ERROR_CODE_LENGTH}), + claim_decision_id INTEGER CHECK(claim_decision_id IS NULL OR claim_decision_id > 0), + CHECK( + (processing_state = 'pending' AND lease_token_sha256 IS NULL + AND lease_expires_at_ms IS NULL AND completed_at_ms IS NULL + AND claim_decision_id IS NULL) + OR + (processing_state = 'processing' AND lease_token_sha256 IS NOT NULL + AND lease_expires_at_ms IS NOT NULL AND completed_at_ms IS NULL + AND claim_decision_id IS NULL) + OR + (processing_state = 'succeeded' AND lease_token_sha256 IS NULL + AND lease_expires_at_ms IS NULL AND completed_at_ms IS NOT NULL + AND claim_decision_id IS NOT NULL) + OR + (processing_state = 'dead_letter' AND lease_token_sha256 IS NULL + AND lease_expires_at_ms IS NULL AND completed_at_ms IS NOT NULL + AND claim_decision_id IS NULL) + ) + ); + + CREATE TABLE IF NOT EXISTS billing_stripe_reconciliation_attempts ( + attempt_id INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT NOT NULL + REFERENCES billing_stripe_reconciliation_jobs(event_id) ON DELETE CASCADE, + attempt_number INTEGER NOT NULL CHECK(attempt_number > 0), + lease_started_at_ms INTEGER NOT NULL CHECK(lease_started_at_ms >= 0), + lease_expires_at_ms INTEGER NOT NULL CHECK(lease_expires_at_ms >= lease_started_at_ms), + finished_at_ms INTEGER CHECK(finished_at_ms IS NULL OR finished_at_ms >= lease_started_at_ms), + outcome TEXT CHECK(outcome IS NULL OR outcome IN ('succeeded','retry','dead_letter')), + error_code TEXT CHECK(error_code IS NULL OR length(error_code) BETWEEN 1 AND ${MAX_ERROR_CODE_LENGTH}), + UNIQUE(event_id, attempt_number) + ); + + CREATE INDEX IF NOT EXISTS billing_stripe_reconciliation_ready_jobs + ON billing_stripe_reconciliation_jobs(processing_state, next_attempt_at_ms, event_id); + CREATE INDEX IF NOT EXISTS billing_stripe_reconciliation_attempt_history + ON billing_stripe_reconciliation_attempts(event_id, attempt_number); + `); +} + +/** + * Create the SQLite repository that leases, retries, completes, and dead-letters work. + * + * Each claim receives an opaque finite lease. The plaintext lease exists only in the + * worker process; SQLite stores its SHA-256 digest. Expired leases are auditable and + * reclaimable until the bounded attempt budget is exhausted, at which point the job + * becomes a durable dead letter instead of remaining in an invisible pending state. + * The 90-second default lease leaves margin beyond the current two sequential + * 15-second authoritative Subscription and Invoice request budgets. + * + * @param {import('node:sqlite').DatabaseSync} database bootstrapped SQLite database + * @param {object} [options] deterministic runtime controls + * @param {() => number} [options.now] wall-clock milliseconds + * @param {() => string} [options.randomToken] opaque lease-token generator + * @param {number} [options.leaseMs=90000] lease lifetime + * @param {number} [options.maxAttempts=5] total attempt budget + * @param {number} [options.baseBackoffMs=5000] first retry delay + * @param {number} [options.maxBackoffMs=300000] retry-delay ceiling + * @returns {Readonly} durable worker repository + */ +export function createSqliteStripeReconciliationWorkerRepository(database, { + now = Date.now, + randomToken = randomUUID, + leaseMs = DEFAULT_LEASE_MS, + maxAttempts = DEFAULT_MAX_ATTEMPTS, + baseBackoffMs = DEFAULT_BASE_BACKOFF_MS, + maxBackoffMs = DEFAULT_MAX_BACKOFF_MS, +} = {}) { + if (!database || typeof database.prepare !== 'function' || typeof database.exec !== 'function') { + throw new TypeError('database must provide SQLite prepare/exec operations'); + } + if (typeof now !== 'function') throw new TypeError('now must be a function'); + if (typeof randomToken !== 'function') throw new TypeError('randomToken must be a function'); + + const leaseDuration = positiveOption(leaseMs, DEFAULT_LEASE_MS, 'leaseMs'); + const attemptBudget = positiveOption(maxAttempts, DEFAULT_MAX_ATTEMPTS, 'maxAttempts'); + const firstBackoff = positiveOption(baseBackoffMs, DEFAULT_BASE_BACKOFF_MS, 'baseBackoffMs'); + const backoffCeiling = positiveOption(maxBackoffMs, DEFAULT_MAX_BACKOFF_MS, 'maxBackoffMs'); + if (firstBackoff > backoffCeiling) { + throw new TypeError('baseBackoffMs must not exceed maxBackoffMs'); + } + + const seedJobs = database.prepare(` + INSERT OR IGNORE INTO billing_stripe_reconciliation_jobs( + event_id, processing_state, attempt_count, next_attempt_at_ms, + lease_token_sha256, lease_expires_at_ms, completed_at_ms, + last_error_code, claim_decision_id + ) + SELECT event_id, 'pending', 0, queued_at_ms, NULL, NULL, NULL, NULL, NULL + FROM billing_stripe_reconciliation_triggers + `); + const selectExpired = database.prepare(` + SELECT event_id, attempt_count + FROM billing_stripe_reconciliation_jobs + WHERE processing_state = 'processing' AND lease_expires_at_ms <= ? + ORDER BY lease_expires_at_ms, event_id + `); + const releaseExpired = database.prepare(` + UPDATE billing_stripe_reconciliation_jobs + SET processing_state = 'pending', next_attempt_at_ms = ?, + lease_token_sha256 = NULL, lease_expires_at_ms = NULL, + last_error_code = ? + WHERE event_id = ? AND processing_state = 'processing' AND lease_expires_at_ms <= ? + `); + const deadLetterExpired = database.prepare(` + UPDATE billing_stripe_reconciliation_jobs + SET processing_state = 'dead_letter', next_attempt_at_ms = ?, + lease_token_sha256 = NULL, lease_expires_at_ms = NULL, + completed_at_ms = ?, last_error_code = ?, claim_decision_id = NULL + WHERE event_id = ? AND processing_state = 'processing' AND lease_expires_at_ms <= ? + `); + const selectReady = database.prepare(` + SELECT jobs.event_id, triggers.subscription_id, jobs.attempt_count + FROM billing_stripe_reconciliation_jobs AS jobs + JOIN billing_stripe_reconciliation_triggers AS triggers USING(event_id) + WHERE jobs.processing_state = 'pending' + AND jobs.next_attempt_at_ms <= ? + AND jobs.attempt_count < ? + ORDER BY jobs.next_attempt_at_ms, triggers.queued_at_ms, jobs.event_id + LIMIT 1 + `); + const claimJob = database.prepare(` + UPDATE billing_stripe_reconciliation_jobs + SET processing_state = 'processing', attempt_count = attempt_count + 1, + lease_token_sha256 = ?, lease_expires_at_ms = ?, last_error_code = NULL + WHERE event_id = ? AND processing_state = 'pending' + AND next_attempt_at_ms <= ? AND attempt_count = ? + `); + const insertAttempt = database.prepare(` + INSERT INTO billing_stripe_reconciliation_attempts( + event_id, attempt_number, lease_started_at_ms, lease_expires_at_ms, + finished_at_ms, outcome, error_code + ) VALUES(?,?,?,?,NULL,NULL,NULL) + `); + const selectLease = database.prepare(` + SELECT attempt_count, lease_token_sha256, lease_expires_at_ms + FROM billing_stripe_reconciliation_jobs + WHERE event_id = ? AND processing_state = 'processing' + `); + const finishAttempt = database.prepare(` + UPDATE billing_stripe_reconciliation_attempts + SET finished_at_ms = ?, outcome = ?, error_code = ? + WHERE event_id = ? AND attempt_number = ? AND outcome IS NULL + `); + const completeJob = database.prepare(` + UPDATE billing_stripe_reconciliation_jobs + SET processing_state = 'succeeded', next_attempt_at_ms = ?, + lease_token_sha256 = NULL, lease_expires_at_ms = NULL, + completed_at_ms = ?, last_error_code = NULL, claim_decision_id = ? + WHERE event_id = ? AND processing_state = 'processing' AND lease_token_sha256 = ? + `); + const failJob = database.prepare(` + UPDATE billing_stripe_reconciliation_jobs + SET processing_state = ?, next_attempt_at_ms = ?, + lease_token_sha256 = NULL, lease_expires_at_ms = NULL, + completed_at_ms = ?, last_error_code = ?, claim_decision_id = NULL + WHERE event_id = ? AND processing_state = 'processing' AND lease_token_sha256 = ? + `); + const selectOrganization = database.prepare(` + SELECT customers.organization_id + FROM billing_stripe_subscriptions AS subscriptions + JOIN billing_stripe_customers AS customers + ON customers.customer_id = subscriptions.customer_id + WHERE subscriptions.subscription_id = ? + `); + + function requireCurrentLease(eventId, leaseToken, nowMs) { + const normalizedEventId = boundedIdentifier(eventId, EVENT_ID_PATTERN); + const leaseHash = tokenHash(leaseToken); + const current = selectLease.get(normalizedEventId); + if (!current + || current.lease_token_sha256 !== leaseHash + || !Number.isSafeInteger(current.lease_expires_at_ms) + || current.lease_expires_at_ms <= nowMs) { + throw workerError('stripe_reconciliation_lease_stale', 409); + } + return { + eventId: normalizedEventId, + attemptNumber: positiveInteger(current.attempt_count, 'attemptNumber'), + leaseHash, + }; + } + + function finishAttemptExactly(nowMs, outcome, errorCode, eventId, attemptNumber) { + const result = finishAttempt.run(nowMs, outcome, errorCode, eventId, attemptNumber); + if (Number(result.changes) !== 1) { + throw workerError('stripe_reconciliation_attempt_state_invalid', 500); + } + } + + return Object.freeze({ + /** Claim at most one ready trigger under an opaque finite lease. */ + claimNext() { + const nowMs = normalizedNow(now); + return withSavepoint(database, () => { + seedJobs.run(); + for (const expired of selectExpired.all(nowMs)) { + const eventId = boundedIdentifier(expired.event_id, EVENT_ID_PATTERN); + const attemptNumber = positiveInteger(expired.attempt_count, 'attemptNumber'); + const terminal = attemptNumber >= attemptBudget; + finishAttemptExactly( + nowMs, + terminal ? 'dead_letter' : 'retry', + LEASE_EXPIRED_CODE, + eventId, + attemptNumber, + ); + const update = terminal + ? deadLetterExpired.run(nowMs, nowMs, LEASE_EXPIRED_CODE, eventId, nowMs) + : releaseExpired.run(nowMs, LEASE_EXPIRED_CODE, eventId, nowMs); + if (Number(update.changes) !== 1) { + throw workerError('stripe_reconciliation_attempt_state_invalid', 500); + } + } + + const candidate = selectReady.get(nowMs, attemptBudget); + if (!candidate) return null; + const eventId = boundedIdentifier(candidate.event_id, EVENT_ID_PATTERN); + const subscriptionId = boundedIdentifier(candidate.subscription_id, SUBSCRIPTION_ID_PATTERN); + const previousAttemptCount = nonNegativeInteger(candidate.attempt_count, 'attemptCount'); + const leaseToken = leaseTokenValue(randomToken); + const leaseHash = tokenHash(leaseToken); + const leaseExpiresAtMs = safeAdd(nowMs, leaseDuration); + const claimed = claimJob.run( + leaseHash, + leaseExpiresAtMs, + eventId, + nowMs, + previousAttemptCount, + ); + if (Number(claimed.changes) !== 1) { + throw workerError('stripe_reconciliation_claim_conflict', 409); + } + const attemptNumber = previousAttemptCount + 1; + insertAttempt.run(eventId, attemptNumber, nowMs, leaseExpiresAtMs); + return Object.freeze({ + eventId, + subscriptionId, + attemptNumber, + leaseToken, + leaseExpiresAtMs, + }); + }); + }, + + /** Resolve tenant authority only from the normalized Subscription→Customer chain. */ + resolveOrganizationId(subscriptionId) { + const id = boundedIdentifier(subscriptionId, SUBSCRIPTION_ID_PATTERN); + const row = selectOrganization.get(id); + return row ? positiveInteger(Number(row.organization_id), 'organizationId') : null; + }, + + /** Complete the exact active lease with one validated durable claim decision ID. */ + complete({ eventId, leaseToken, claimDecisionId } = {}) { + const nowMs = normalizedNow(now); + const decisionId = positiveInteger(claimDecisionId, 'claimDecisionId'); + return withSavepoint(database, () => { + const lease = requireCurrentLease(eventId, leaseToken, nowMs); + const updated = completeJob.run( + nowMs, + nowMs, + decisionId, + lease.eventId, + lease.leaseHash, + ); + if (Number(updated.changes) !== 1) { + throw workerError('stripe_reconciliation_lease_stale', 409); + } + finishAttemptExactly(nowMs, 'succeeded', null, lease.eventId, lease.attemptNumber); + return Object.freeze({ + eventId: lease.eventId, + status: 'succeeded', + claimDecisionId: decisionId, + }); + }); + }, + + /** Record a sanitized retry or terminal dead letter for the exact active lease. */ + fail({ eventId, leaseToken, errorCode } = {}) { + const nowMs = normalizedNow(now); + const code = boundedFailureCode(errorCode); + return withSavepoint(database, () => { + const lease = requireCurrentLease(eventId, leaseToken, nowMs); + const terminal = lease.attemptNumber >= attemptBudget; + const state = terminal ? 'dead_letter' : 'pending'; + const outcome = terminal ? 'dead_letter' : 'retry'; + const nextAttemptAtMs = terminal + ? nowMs + : safeAdd(nowMs, retryDelay(lease.attemptNumber, firstBackoff, backoffCeiling)); + const updated = failJob.run( + state, + nextAttemptAtMs, + terminal ? nowMs : null, + code, + lease.eventId, + lease.leaseHash, + ); + if (Number(updated.changes) !== 1) { + throw workerError('stripe_reconciliation_lease_stale', 409); + } + finishAttemptExactly(nowMs, outcome, code, lease.eventId, lease.attemptNumber); + return Object.freeze({ + eventId: lease.eventId, + status: terminal ? 'dead_letter' : 'retry', + errorCode: code, + nextAttemptAtMs: terminal ? null : nextAttemptAtMs, + }); + }); + }, + }); +} + +/** + * Consume at most one queued Stripe reconciliation trigger. + * + * The worker derives the organization from server-owned normalized Stripe identity, + * then invokes the authoritative reconciliation service, which re-fetches current + * provider state before producing a claim decision. A receipt must remain bound to + * the claimed tenant and Subscription. Missing identity and causal failures become + * bounded retry/dead-letter evidence, while arbitrary exception text is never stored. + * Once reconciliation has succeeded, a completion failure is treated as uncertain + * durable state and must not initiate a contradictory failure transition. + * + * @param {object} input worker orchestration ports + * @param {object} input.repository durable worker repository + * @param {Function} input.reconcile authoritative billing reconciliation function + * @param {object} [input.reconciliationDependencies] server-owned dependency ports/options + * @returns {Promise>} idle, retry/dead-letter, or success receipt + */ +export async function runNextStripeReconciliationJob({ + repository, + reconcile, + reconciliationDependencies = {}, +}) { + if (!repository + || typeof repository.claimNext !== 'function' + || typeof repository.resolveOrganizationId !== 'function' + || typeof repository.complete !== 'function' + || typeof repository.fail !== 'function') { + throw new TypeError('repository must provide claim, authority, completion, and failure operations'); + } + if (typeof reconcile !== 'function') throw new TypeError('reconcile must be a function'); + if (!reconciliationDependencies + || typeof reconciliationDependencies !== 'object' + || Array.isArray(reconciliationDependencies)) { + throw new TypeError('reconciliationDependencies must be an object'); + } + + const claim = repository.claimNext(); + if (claim == null) return Object.freeze({ status: 'idle' }); + + const organizationId = repository.resolveOrganizationId(claim.subscriptionId); + if (organizationId == null) { + return repository.fail({ + eventId: claim.eventId, + leaseToken: claim.leaseToken, + errorCode: 'stripe_reconciliation_authority_missing', + }); + } + + let claimDecisionId; + try { + const receipt = await reconcile({ + ...reconciliationDependencies, + organizationId, + subscriptionId: claim.subscriptionId, + sourceEventId: claim.eventId, + }); + if (!receipt + || typeof receipt !== 'object' + || Array.isArray(receipt) + || receipt.organizationId !== organizationId + || receipt.subscriptionId !== claim.subscriptionId) { + throw workerError('stripe_reconciliation_receipt_mismatch', 500); + } + claimDecisionId = positiveInteger(receipt.claimDecisionId, 'claimDecisionId'); + } catch (error) { + const failure = repository.fail({ + eventId: claim.eventId, + leaseToken: claim.leaseToken, + errorCode: safeFailureCode(error), + }); + return Object.freeze({ + ...failure, + subscriptionId: claim.subscriptionId, + organizationId, + }); + } + + try { + repository.complete({ + eventId: claim.eventId, + leaseToken: claim.leaseToken, + claimDecisionId, + }); + } catch { + throw workerError('stripe_reconciliation_worker_state_uncertain', 500); + } + + return Object.freeze({ + status: 'succeeded', + eventId: claim.eventId, + subscriptionId: claim.subscriptionId, + organizationId, + claimDecisionId, + }); +} diff --git a/tests/unit/coverage-script-contract.test.mjs b/tests/unit/coverage-script-contract.test.mjs index 3b57d82e..24ba9a48 100644 --- a/tests/unit/coverage-script-contract.test.mjs +++ b/tests/unit/coverage-script-contract.test.mjs @@ -69,6 +69,11 @@ assert.match( /--include=server\/stripe_checkout_identity_bootstrap\.mjs/, 'the verified Checkout identity bootstrap boundary is instrumented', ); +assert.match( + scripts['test:coverage'], + /--include=server\/stripe_reconciliation_worker\.mjs/, + 'the leased Stripe reconciliation worker is instrumented', +); assert.match( scripts['test:coverage'], /--include=server\/stripe_subscription_provider\.mjs/, @@ -169,6 +174,11 @@ assert.match( /tests\/unit\/stripe-webhook-reconciliation-queue-integration\.test\.mjs/, 'the verified webhook-to-queue integration executes under c8', ); +assert.match( + scripts['test:coverage:cases'], + /tests\/unit\/stripe-reconciliation-worker\.test\.mjs/, + 'the leased Stripe reconciliation worker executes under c8', +); assert.match( scripts['test:unit'], /tests\/unit\/stripe-webhook-reconciliation-queue\.test\.mjs/, @@ -184,6 +194,11 @@ assert.match( /tests\/unit\/stripe-webhook-reconciliation-queue-integration\.test\.mjs/, 'normal unit CI executes the verified webhook-to-queue integration', ); +assert.match( + scripts['test:unit'], + /tests\/unit\/stripe-reconciliation-worker\.test\.mjs/, + 'normal unit CI executes the leased Stripe reconciliation worker regression', +); assert.match( scripts['test:coverage:cases'], /tests\/unit\/stripe-subscription-provider\.test\.mjs/, diff --git a/tests/unit/stripe-reconciliation-worker-time-budget.test.mjs b/tests/unit/stripe-reconciliation-worker-time-budget.test.mjs new file mode 100644 index 00000000..a355e498 --- /dev/null +++ b/tests/unit/stripe-reconciliation-worker-time-budget.test.mjs @@ -0,0 +1,117 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { DatabaseSync } from 'node:sqlite'; + +import { installStripeSubscriptionObservationSchema } from '../../server/stripe_subscription_observation_ledger.mjs'; +import { installStripeWebhookEventSchema } from '../../server/stripe_webhook_event_ledger.mjs'; +import { installStripeWebhookReconciliationQueueSchema } from '../../server/stripe_webhook_reconciliation_queue.mjs'; +import { + createSqliteStripeReconciliationWorkerRepository, + installStripeReconciliationWorkerSchema, + runNextStripeReconciliationJob, +} from '../../server/stripe_reconciliation_worker.mjs'; + +function databaseWithReadyTrigger() { + const database = new DatabaseSync(':memory:'); + database.exec('PRAGMA foreign_keys = ON'); + database.exec(` + CREATE TABLE orgs ( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL, + plan TEXT NOT NULL DEFAULT 'free' + ); + `); + installStripeWebhookEventSchema(database); + installStripeSubscriptionObservationSchema(database); + installStripeWebhookReconciliationQueueSchema(database); + installStripeReconciliationWorkerSchema(database); + database.prepare('INSERT INTO orgs(id,name,plan) VALUES(?,?,?)').run(7, 'Lease Budget Org', 'free'); + database.prepare(` + INSERT INTO billing_stripe_customers(customer_id, organization_id, first_observed_at_ms) + VALUES(?,?,?) + `).run('cus_lease_budget', 7, 1_000); + database.prepare(` + INSERT INTO billing_stripe_subscriptions(subscription_id, customer_id, first_observed_at_ms) + VALUES(?,?,?) + `).run('sub_lease_budget', 'cus_lease_budget', 1_000); + database.prepare(` + INSERT INTO billing_stripe_webhook_events( + event_id, provider_created_at_sec, event_type, object_id, object_type, + api_version, request_id, payload_sha256, first_received_at_ms + ) VALUES(?,?,?,?,?,?,?,?,?) + `).run( + 'evt_lease_budget', + 1_787_000_000, + 'customer.subscription.updated', + 'sub_lease_budget', + 'subscription', + '2025-03-31.basil', + null, + 'b'.repeat(64), + 1_000, + ); + database.prepare(` + INSERT INTO billing_stripe_reconciliation_triggers( + event_id, subscription_id, queued_at_ms, processing_state + ) VALUES(?,?,?,'pending') + `).run('evt_lease_budget', 'sub_lease_budget', 1_000); + return database; +} + +test('default worker lease exceeds the two sequential 15-second authoritative provider budgets', () => { + const database = databaseWithReadyTrigger(); + const nowMs = 2_000; + const repository = createSqliteStripeReconciliationWorkerRepository(database, { + now: () => nowMs, + randomToken: () => 'lease_token_budget_1234567890', + }); + + const claim = repository.claimNext(); + assert.ok(claim); + assert.ok( + claim.leaseExpiresAtMs - nowMs > 30_000, + 'default lease must leave completion margin beyond Subscription + Invoice timeout budgets', + ); +}); + +test('provider success followed by uncertain worker completion never starts a failure transition', async () => { + let failureTransitions = 0; + const repository = { + claimNext() { + return { + eventId: 'evt_worker_completion_uncertain', + subscriptionId: 'sub_worker_completion_uncertain', + leaseToken: 'lease_token_completion_123456', + }; + }, + resolveOrganizationId() { + return 7; + }, + complete() { + throw new Error('injected completion state uncertainty'); + }, + fail() { + failureTransitions += 1; + return { + status: 'retry', + eventId: 'evt_worker_completion_uncertain', + errorCode: 'stripe_reconciliation_failed', + nextAttemptAtMs: 9_000, + }; + }, + }; + + await assert.rejects( + runNextStripeReconciliationJob({ + repository, + reconcile: async () => ({ + organizationId: 7, + subscriptionId: 'sub_worker_completion_uncertain', + claimDecisionId: 91, + }), + }), + (error) => error?.code === 'stripe_reconciliation_worker_state_uncertain' + && error.status === 500, + ); + assert.equal(failureTransitions, 0); +}); diff --git a/tests/unit/stripe-reconciliation-worker.test.mjs b/tests/unit/stripe-reconciliation-worker.test.mjs new file mode 100644 index 00000000..e2c3f5a7 --- /dev/null +++ b/tests/unit/stripe-reconciliation-worker.test.mjs @@ -0,0 +1,317 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { DatabaseSync } from 'node:sqlite'; + +import { installStripeSubscriptionObservationSchema } from '../../server/stripe_subscription_observation_ledger.mjs'; +import { installStripeWebhookEventSchema } from '../../server/stripe_webhook_event_ledger.mjs'; +import { installStripeWebhookReconciliationQueueSchema } from '../../server/stripe_webhook_reconciliation_queue.mjs'; +import { + StripeReconciliationWorkerError, + createSqliteStripeReconciliationWorkerRepository, + installStripeReconciliationWorkerSchema, + runNextStripeReconciliationJob, +} from '../../server/stripe_reconciliation_worker.mjs'; + +function createWorkerDatabase() { + const database = new DatabaseSync(':memory:'); + database.exec('PRAGMA foreign_keys = ON'); + database.exec(` + CREATE TABLE orgs ( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL, + plan TEXT NOT NULL DEFAULT 'free' + ); + `); + installStripeWebhookEventSchema(database); + installStripeSubscriptionObservationSchema(database); + installStripeWebhookReconciliationQueueSchema(database); + installStripeReconciliationWorkerSchema(database); + return database; +} + +function seedTrigger(database, { + eventId = 'evt_worker', + subscriptionId = 'sub_worker', + organizationId = 7, + withAuthority = true, +} = {}) { + database.prepare('INSERT INTO orgs(id, name, plan) VALUES(?,?,?)') + .run(organizationId, 'Worker Org', 'free'); + if (withAuthority) { + database.prepare(` + INSERT INTO billing_stripe_customers(customer_id, organization_id, first_observed_at_ms) + VALUES(?,?,?) + `).run('cus_worker', organizationId, 1_000); + database.prepare(` + INSERT INTO billing_stripe_subscriptions(subscription_id, customer_id, first_observed_at_ms) + VALUES(?,?,?) + `).run(subscriptionId, 'cus_worker', 1_000); + } + database.prepare(` + INSERT INTO billing_stripe_webhook_events( + event_id, provider_created_at_sec, event_type, object_id, object_type, + api_version, request_id, payload_sha256, first_received_at_ms + ) VALUES(?,?,?,?,?,?,?,?,?) + `).run( + eventId, + 1_787_000_000, + 'customer.subscription.updated', + subscriptionId, + 'subscription', + '2025-03-31.basil', + null, + 'a'.repeat(64), + 1_000, + ); + database.prepare(` + INSERT INTO billing_stripe_reconciliation_triggers( + event_id, subscription_id, queued_at_ms, processing_state + ) VALUES(?,?,?,'pending') + `).run(eventId, subscriptionId, 1_000); +} + +function sequentialTokens() { + let index = 0; + return () => `lease_token_${String(++index).padStart(16, '0')}`; +} + +test('worker claims one durable trigger, uses server-owned tenant authority, and records successful completion', async () => { + const database = createWorkerDatabase(); + seedTrigger(database); + let nowMs = 2_000; + const repository = createSqliteStripeReconciliationWorkerRepository(database, { + now: () => nowMs, + randomToken: sequentialTokens(), + leaseMs: 30_000, + }); + const calls = []; + + const result = await runNextStripeReconciliationJob({ + repository, + reconcile: async (input) => { + calls.push(input); + return { + organizationId: input.organizationId, + subscriptionId: input.subscriptionId, + subscriptionObservationId: 11, + invoiceObservationId: 12, + claimDecisionId: 13, + }; + }, + reconciliationDependencies: { secretKey: 'sk_test_server_owned' }, + }); + + assert.equal(result.status, 'succeeded'); + assert.equal(result.eventId, 'evt_worker'); + assert.equal(result.subscriptionId, 'sub_worker'); + assert.equal(result.organizationId, 7); + assert.equal(result.claimDecisionId, 13); + assert.deepEqual(calls, [{ + organizationId: 7, + subscriptionId: 'sub_worker', + sourceEventId: 'evt_worker', + secretKey: 'sk_test_server_owned', + }]); + + const job = database.prepare(` + SELECT processing_state, attempt_count, claim_decision_id, lease_token_sha256, + lease_expires_at_ms, completed_at_ms, last_error_code + FROM billing_stripe_reconciliation_jobs WHERE event_id = ? + `).get('evt_worker'); + assert.deepEqual({ ...job }, { + processing_state: 'succeeded', + attempt_count: 1, + claim_decision_id: 13, + lease_token_sha256: null, + lease_expires_at_ms: null, + completed_at_ms: nowMs, + last_error_code: null, + }); + + const attempt = database.prepare(` + SELECT attempt_number, outcome, error_code + FROM billing_stripe_reconciliation_attempts WHERE event_id = ? + `).get('evt_worker'); + assert.deepEqual({ ...attempt }, { + attempt_number: 1, + outcome: 'succeeded', + error_code: null, + }); + assert.equal(repository.claimNext(), null, 'completed work is never claimed again'); +}); + +test('leases prevent concurrent duplicate processing and stale workers cannot complete reclaimed work', () => { + const database = createWorkerDatabase(); + seedTrigger(database); + let nowMs = 5_000; + const repository = createSqliteStripeReconciliationWorkerRepository(database, { + now: () => nowMs, + randomToken: sequentialTokens(), + leaseMs: 100, + }); + + const first = repository.claimNext(); + assert.equal(first.attemptNumber, 1); + assert.equal(repository.claimNext(), null, 'an unexpired lease excludes another worker'); + + nowMs = 5_101; + const second = repository.claimNext(); + assert.equal(second.attemptNumber, 2); + assert.notEqual(second.leaseToken, first.leaseToken); + + assert.throws( + () => repository.complete({ + eventId: first.eventId, + leaseToken: first.leaseToken, + claimDecisionId: 41, + }), + (error) => error instanceof StripeReconciliationWorkerError + && error.code === 'stripe_reconciliation_lease_stale', + ); + + repository.complete({ + eventId: second.eventId, + leaseToken: second.leaseToken, + claimDecisionId: 42, + }); + const attempts = database.prepare(` + SELECT attempt_number, outcome, error_code + FROM billing_stripe_reconciliation_attempts + WHERE event_id = ? ORDER BY attempt_number + `).all('evt_worker').map((row) => ({ ...row })); + assert.deepEqual(attempts, [ + { + attempt_number: 1, + outcome: 'retry', + error_code: 'stripe_reconciliation_lease_expired', + }, + { + attempt_number: 2, + outcome: 'succeeded', + error_code: null, + }, + ]); +}); + +test('worker failures back off, dead-letter at the bounded attempt budget, and never persist arbitrary provider text', async () => { + const database = createWorkerDatabase(); + seedTrigger(database); + let nowMs = 10_000; + const repository = createSqliteStripeReconciliationWorkerRepository(database, { + now: () => nowMs, + randomToken: sequentialTokens(), + leaseMs: 1_000, + maxAttempts: 2, + baseBackoffMs: 50, + maxBackoffMs: 500, + }); + const unsafeMessage = 'provider failed with sk_live_should_never_be_persisted'; + + const first = await runNextStripeReconciliationJob({ + repository, + reconcile: async () => { + throw new Error(unsafeMessage); + }, + }); + assert.equal(first.status, 'retry'); + assert.equal(first.errorCode, 'stripe_reconciliation_failed'); + assert.equal(repository.claimNext(), null, 'backoff prevents an immediate hot loop'); + + nowMs += 50; + const second = await runNextStripeReconciliationJob({ + repository, + reconcile: async () => { + throw new Error(unsafeMessage); + }, + }); + assert.equal(second.status, 'dead_letter'); + assert.equal(second.errorCode, 'stripe_reconciliation_failed'); + + const persisted = database.prepare(` + SELECT processing_state, attempt_count, last_error_code + FROM billing_stripe_reconciliation_jobs WHERE event_id = ? + `).get('evt_worker'); + assert.deepEqual({ ...persisted }, { + processing_state: 'dead_letter', + attempt_count: 2, + last_error_code: 'stripe_reconciliation_failed', + }); + const persistedText = JSON.stringify(database.prepare(` + SELECT error_code FROM billing_stripe_reconciliation_attempts WHERE event_id = ? + `).all('evt_worker')); + assert.equal(persistedText.includes('sk_live'), false); + assert.equal(repository.claimNext(), null); +}); + +test('missing tenant identity remains explicit retryable work and never calls the provider reconciliation port', async () => { + const database = createWorkerDatabase(); + seedTrigger(database, { withAuthority: false }); + let nowMs = 20_000; + const repository = createSqliteStripeReconciliationWorkerRepository(database, { + now: () => nowMs, + randomToken: sequentialTokens(), + baseBackoffMs: 25, + }); + let reconcileCalls = 0; + + const result = await runNextStripeReconciliationJob({ + repository, + reconcile: async () => { + reconcileCalls += 1; + throw new Error('must not run'); + }, + }); + + assert.equal(reconcileCalls, 0); + assert.equal(result.status, 'retry'); + assert.equal(result.errorCode, 'stripe_reconciliation_authority_missing'); + const job = database.prepare(` + SELECT processing_state, next_attempt_at_ms, last_error_code + FROM billing_stripe_reconciliation_jobs WHERE event_id = ? + `).get('evt_worker'); + assert.equal(job.processing_state, 'pending'); + assert.equal(job.next_attempt_at_ms, nowMs + 25); + assert.equal(job.last_error_code, 'stripe_reconciliation_authority_missing'); +}); + +test('dependency options cannot override server-owned tenant, Subscription, or verified Event authority', async () => { + const database = createWorkerDatabase(); + seedTrigger(database, { + eventId: 'evt_authority_worker', + subscriptionId: 'sub_authority_worker', + organizationId: 17, + }); + const repository = createSqliteStripeReconciliationWorkerRepository(database, { + now: () => 30_000, + randomToken: sequentialTokens(), + }); + const calls = []; + + const result = await runNextStripeReconciliationJob({ + repository, + reconcile: async (input) => { + calls.push(input); + return { + organizationId: input.organizationId, + subscriptionId: input.subscriptionId, + claimDecisionId: 71, + }; + }, + reconciliationDependencies: { + organizationId: 999, + subscriptionId: 'sub_foreign_override', + sourceEventId: 'evt_foreign_override', + secretKey: 'sk_test_server_owned', + }, + }); + + assert.equal(result.status, 'succeeded'); + assert.equal(result.organizationId, 17); + assert.equal(result.subscriptionId, 'sub_authority_worker'); + assert.deepEqual(calls, [{ + organizationId: 17, + subscriptionId: 'sub_authority_worker', + sourceEventId: 'evt_authority_worker', + secretKey: 'sk_test_server_owned', + }]); +});