feat: implement idempotent Stellar wallet provisioning

prisma/migrations/20260818090000_idempotent_wallet_provisioning/migration.sql

@@ -0,0 +1,56 @@
+-- Wallet provisioning stores public Stellar material and opaque KMS references only.
+CREATE TABLE "managed_key_references" (
+    "id" TEXT NOT NULL,
+    "provider" TEXT NOT NULL,
+    "opaqueReference" TEXT NOT NULL,
+    "keyVersion" TEXT,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+    CONSTRAINT "managed_key_references_pkey" PRIMARY KEY ("id")
+);
+
+CREATE TABLE "wallets" (
+    "id" TEXT NOT NULL,
+    "userId" TEXT NOT NULL,
+    "network" TEXT NOT NULL,
+    "custody" TEXT NOT NULL DEFAULT 'MANAGED',
+    "publicKey" TEXT,
+    "status" TEXT NOT NULL DEFAULT 'RESERVED',
+    "managedKeyReferenceId" TEXT,
+    "failureCode" TEXT,
+    "attemptCount" INTEGER NOT NULL DEFAULT 0,
+    "provisionedAt" TIMESTAMP(3),
+    "statusChangedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+    CONSTRAINT "wallets_pkey" PRIMARY KEY ("id")
+);
+
+CREATE TABLE "wallet_provisioning_jobs" (
+    "id" TEXT NOT NULL,
+    "walletId" TEXT NOT NULL,
+    "status" TEXT NOT NULL DEFAULT 'PENDING',
+    "availableAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "leaseToken" TEXT,
+    "leasedUntil" TIMESTAMP(3),
+    "attempts" INTEGER NOT NULL DEFAULT 0,
+    "maxAttempts" INTEGER NOT NULL DEFAULT 8,
+    "lastFailureCode" TEXT,
+    "completedAt" TIMESTAMP(3),
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+    CONSTRAINT "wallet_provisioning_jobs_pkey" PRIMARY KEY ("id")
+);
+
+CREATE UNIQUE INDEX "managed_key_references_opaqueReference_key" ON "managed_key_references"("opaqueReference");
+CREATE UNIQUE INDEX "wallets_userId_key" ON "wallets"("userId");
+CREATE UNIQUE INDEX "wallets_publicKey_key" ON "wallets"("publicKey");
+CREATE UNIQUE INDEX "wallets_managedKeyReferenceId_key" ON "wallets"("managedKeyReferenceId");
+CREATE INDEX "wallets_status_updatedAt_idx" ON "wallets"("status", "updatedAt");
+CREATE UNIQUE INDEX "wallet_provisioning_jobs_walletId_key" ON "wallet_provisioning_jobs"("walletId");
+CREATE INDEX "wallet_provisioning_jobs_status_availableAt_idx" ON "wallet_provisioning_jobs"("status", "availableAt");
+CREATE INDEX "wallet_provisioning_jobs_status_leasedUntil_idx" ON "wallet_provisioning_jobs"("status", "leasedUntil");
+
+ALTER TABLE "wallets" ADD CONSTRAINT "wallets_userId_fkey" FOREIGN KEY ("userId") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+ALTER TABLE "wallets" ADD CONSTRAINT "wallets_managedKeyReferenceId_fkey" FOREIGN KEY ("managedKeyReferenceId") REFERENCES "managed_key_references"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+ALTER TABLE "wallet_provisioning_jobs" ADD CONSTRAINT "wallet_provisioning_jobs_walletId_fkey" FOREIGN KEY ("walletId") REFERENCES "wallets"("id") ON DELETE CASCADE ON UPDATE CASCADE;

prisma/schema.prisma

@@ -43,6 +43,7 @@ model User {
   otpChallenges   OtpChallenge[]
   onboarding      OnboardingProgress?
   consentRecords  ConsentRecord[]
+  wallet          Wallet?
 
   @@map("users")
 }
@@ -130,7 +131,7 @@ model ConsentRecord {
   userId        String
   user          User     @relation(fields: [userId], references: [id], onDelete: Cascade)
 
-  purpose       String   // terms_of_service, privacy_policy, marketing_emails, analytics, data_sharing
+  purpose       String   // terms_of_service, privacy_policy, marketing_emails, analytics, data_sharing, custodial_wallet
   required      Boolean  @default(false)
   policyVersion String
   status        String   // granted, withdrawn
@@ -485,3 +486,59 @@ model AccountDeletionRequest {
   @@index([status, scheduledFor])
   @@map("account_deletion_requests")
 }
+
+model ManagedKeyReference {
+  id              String   @id @default(uuid())
+  provider        String
+  opaqueReference String   @unique
+  keyVersion      String?
+  createdAt       DateTime @default(now())
+  updatedAt       DateTime @updatedAt
+
+  wallet Wallet?
+
+  @@map("managed_key_references")
+}
+
+model Wallet {
+  id                    String               @id @default(uuid())
+  userId                String               @unique
+  user                  User                 @relation(fields: [userId], references: [id], onDelete: Cascade)
+  network               String
+  custody               String               @default("MANAGED")
+  publicKey             String?              @unique
+  status                String               @default("RESERVED")
+  managedKeyReferenceId String?              @unique
+  managedKeyReference   ManagedKeyReference? @relation(fields: [managedKeyReferenceId], references: [id], onDelete: Restrict)
+  failureCode           String?
+  attemptCount          Int                  @default(0)
+  provisionedAt         DateTime?
+  statusChangedAt       DateTime             @default(now())
+  createdAt             DateTime             @default(now())
+  updatedAt             DateTime             @updatedAt
+
+  provisioningJob WalletProvisioningJob?
+
+  @@index([status, updatedAt])
+  @@map("wallets")
+}
+
+model WalletProvisioningJob {
+  id              String    @id @default(uuid())
+  walletId        String    @unique
+  wallet          Wallet    @relation(fields: [walletId], references: [id], onDelete: Cascade)
+  status          String    @default("PENDING")
+  availableAt     DateTime  @default(now())
+  leaseToken      String?
+  leasedUntil     DateTime?
+  attempts        Int       @default(0)
+  maxAttempts     Int       @default(8)
+  lastFailureCode String?
+  completedAt     DateTime?
+  createdAt       DateTime  @default(now())
+  updatedAt       DateTime  @updatedAt
+
+  @@index([status, availableAt])
+  @@index([status, leasedUntil])
+  @@map("wallet_provisioning_jobs")
+}

src/jobs/wallet-provisioning.handler.ts

@@ -0,0 +1,142 @@
+import type {
+  KmsSecretStore,
+  StoredStellarKey,
+} from '../services/kms/kms-secret-store'
+import type { StellarKeypairGenerator } from '../services/stellar-keypair.adapter'
+import type { WalletProvisioningRepository } from '../services/wallet-provisioning.repository'
+import type {
+  ClaimedWalletProvisioningJob,
+  WalletProvisioningFailureCode,
+} from '../types/wallet-provisioning.types'
+
+export type WalletProvisioningHandleResult =
+  | { kind: 'idle' }
+  | { kind: 'completed'; walletId: string; publicKey: string }
+  | {
+      kind: 'retry-scheduled'
+      walletId: string
+      failureCode: WalletProvisioningFailureCode
+    }
+  | {
+      kind: 'dead-letter'
+      walletId: string
+      failureCode: WalletProvisioningFailureCode
+    }
+
+export interface WalletProvisioningHandlerOptions {
+  leaseMs?: number
+  baseRetryMs?: number
+  now?: () => Date
+}
+
+/** Durable outbox/job handler for one idempotent wallet provisioning attempt. */
+export class WalletProvisioningOutboxHandler {
+  private readonly leaseMs: number
+  private readonly baseRetryMs: number
+  private readonly now: () => Date
+
+  constructor(
+    private readonly repository: WalletProvisioningRepository,
+    private readonly kms: KmsSecretStore,
+    private readonly keypairGenerator: StellarKeypairGenerator,
+    options: WalletProvisioningHandlerOptions = {},
+  ) {
+    this.leaseMs = options.leaseMs ?? 60_000
+    this.baseRetryMs = options.baseRetryMs ?? 1_000
+    this.now = options.now ?? (() => new Date())
+  }
+
+  async handleNext(): Promise<WalletProvisioningHandleResult> {
+    const claimed = await this.repository.claimNext(this.now(), this.leaseMs)
+    if (!claimed) return { kind: 'idle' }
+
+    const leaseToken = claimed.job.leaseToken
+    if (!leaseToken) {
+      return this.fail(claimed, 'DATABASE_FINALIZATION_FAILED')
+    }
+
+    let material: StoredStellarKey | null
+    try {
+      material = await this.kms.findByIdempotencyKey(claimed.wallet.id)
+    } catch {
+      return this.fail(claimed, 'KMS_UNAVAILABLE')
+    }
+
+    if (!material) {
+      let generated
+      try {
+        generated = this.keypairGenerator.generate()
+      } catch {
+        return this.fail(claimed, 'KEY_GENERATION_FAILED')
+      }
+
+      try {
+        material = await this.kms.storeStellarSecret({
+          idempotencyKey: claimed.wallet.id,
+          publicKey: generated.publicKey,
+          secret: generated.secret,
+        })
+      } catch {
+        // The write may have committed before the provider response was lost.
+        // A deterministic lookup repairs that ambiguity without a second keypair.
+        try {
+          material = await this.kms.findByIdempotencyKey(claimed.wallet.id)
+        } catch {
+          material = null
+        }
+        if (!material) return this.fail(claimed, 'KMS_STORE_UNCERTAIN')
+      }
+    }
+
+    try {
+      const wallet = await this.repository.complete(
+        claimed.wallet.id,
+        leaseToken,
+        material,
+        this.now(),
+      )
+
+      return {
+        kind: 'completed',
+        walletId: wallet.id,
+        publicKey: material.publicKey,
+      }
+    } catch {
+      return this.fail(claimed, 'DATABASE_FINALIZATION_FAILED')
+    }
+  }
+
+  private async fail(
+    claimed: ClaimedWalletProvisioningJob,
+    failureCode: WalletProvisioningFailureCode,
+  ): Promise<WalletProvisioningHandleResult> {
+    const leaseToken = claimed.job.leaseToken
+    const terminal = claimed.job.attempts >= claimed.job.maxAttempts
+    const retryAt = terminal
+      ? null
+      : new Date(
+          this.now().getTime() +
+            this.baseRetryMs *
+              Math.pow(2, Math.max(0, claimed.job.attempts - 1)),
+        )
+
+    if (leaseToken) {
+      try {
+        await this.repository.recordFailure(
+          claimed.wallet.id,
+          leaseToken,
+          failureCode,
+          retryAt,
+          this.now(),
+        )
+      } catch {
+        // A database outage leaves the lease to expire; the job then becomes
+        // claimable again without losing the KMS idempotency key.
+      }
+    }
+
+    return terminal
+      ? { kind: 'dead-letter', walletId: claimed.wallet.id, failureCode }
+      : { kind: 'retry-scheduled', walletId: claimed.wallet.id, failureCode }
+  }
+}

src/services/kms/in-memory-envelope-kms.ts

@@ -0,0 +1,80 @@
+import { createCipheriv, randomBytes, randomUUID } from 'node:crypto'
+import type {
+  KmsSecretStore,
+  StoredStellarKey,
+  StoreStellarSecretInput,
+} from './kms-secret-store'
+
+interface EncryptedEntry {
+  material: StoredStellarKey
+  ciphertext: string
+  iv: string
+  authTag: string
+}
+
+export interface InMemoryEnvelopeKmsHooks {
+  beforeLookup?: () => void | Promise<void>
+  beforeStore?: () => void | Promise<void>
+  afterStore?: () => void | Promise<void>
+}
+
+/**
+ * Development/test KMS adapter. Even this fake keeps only AES-GCM ciphertext,
+ * so snapshots and failure diagnostics cannot accidentally contain a seed.
+ */
+export class InMemoryEnvelopeKms implements KmsSecretStore {
+  readonly #masterKey = randomBytes(32)
+  readonly #entries = new Map<string, EncryptedEntry>()
+
+  constructor(private readonly hooks: InMemoryEnvelopeKmsHooks = {}) {}
+
+  async findByIdempotencyKey(
+    idempotencyKey: string,
+  ): Promise<StoredStellarKey | null> {
+    await this.hooks.beforeLookup?.()
+
+    return this.#entries.get(idempotencyKey)?.material ?? null
+  }
+
+  async storeStellarSecret(
+    input: StoreStellarSecretInput,
+  ): Promise<StoredStellarKey> {
+    const existing = this.#entries.get(input.idempotencyKey)
+    if (existing) return existing.material
+
+    await this.hooks.beforeStore?.()
+
+    const iv = randomBytes(12)
+    const encrypted = input.secret.use((value) => {
+      const cipher = createCipheriv('aes-256-gcm', this.#masterKey, iv)
+      const ciphertext = Buffer.concat([
+        cipher.update(value, 'utf8'),
+        cipher.final(),
+      ])
+
+      return { ciphertext, authTag: cipher.getAuthTag() }
+    })
+
+    const material: StoredStellarKey = {
+      provider: 'in-memory-envelope-kms',
+      opaqueReference: `kms-memory://${randomUUID()}`,
+      keyVersion: '1',
+      publicKey: input.publicKey,
+    }
+
+    this.#entries.set(input.idempotencyKey, {
+      material,
+      ciphertext: encrypted.ciphertext.toString('base64'),
+      iv: iv.toString('base64'),
+      authTag: encrypted.authTag.toString('base64'),
+    })
+
+    await this.hooks.afterStore?.()
+
+    return material
+  }
+
+  get storedKeyCount(): number {
+    return this.#entries.size
+  }
+}

src/services/kms/index.ts

@@ -0,0 +1,2 @@
+export * from './kms-secret-store'
+export * from './in-memory-envelope-kms'