From cb3227b1cacc4328b733927b674cfb8639580119 Mon Sep 17 00:00:00 2001 From: Sadeequ <70214653+Sadeequ@users.noreply.github.com> Date: Wed, 29 Jul 2026 10:05:38 +0000 Subject: [PATCH] Added a full guarded notification for all finacial transaction notification --- Backend/DOCUMENTATION/outbox-changes.md | 122 +++++ Backend/README.md | 87 ++++ Backend/package-lock.json | 16 +- Backend/prisma/schema.prisma | 453 ++++++++++-------- Backend/src/jobs/index.ts | 21 +- Backend/src/jobs/outbox.processor.ts | 200 ++++++++ Backend/src/queues/index.ts | 2 + Backend/src/repositories/outbox.repository.ts | 104 ++++ Backend/src/repositories/user.repository.ts | 13 + Backend/src/server.ts | 2 + Backend/src/services/auth.service.ts | 71 +-- Backend/src/services/notification.service.ts | 101 ++++ Backend/src/services/payment.service.ts | 110 +++-- Backend/src/services/vault.service.ts | 257 ++++++---- .../integration/outbox.integration.test.ts | 288 +++++++++++ Backend/tests/unit/outbox.repository.test.ts | 243 ++++++++++ 16 files changed, 1732 insertions(+), 358 deletions(-) create mode 100644 Backend/DOCUMENTATION/outbox-changes.md create mode 100644 Backend/src/jobs/outbox.processor.ts create mode 100644 Backend/src/repositories/outbox.repository.ts create mode 100644 Backend/src/services/notification.service.ts create mode 100644 Backend/tests/integration/outbox.integration.test.ts create mode 100644 Backend/tests/unit/outbox.repository.test.ts diff --git a/Backend/DOCUMENTATION/outbox-changes.md b/Backend/DOCUMENTATION/outbox-changes.md new file mode 100644 index 0000000..0e0c547 --- /dev/null +++ b/Backend/DOCUMENTATION/outbox-changes.md @@ -0,0 +1,122 @@ +# Transactional Outbox Implementation — Changes Documentation + +## Overview + +This document explains the changes made to implement a transactional outbox pattern for reliable financial and notification job processing in the Vaulty backend. + +## Problem Statement + +Previously, the backend used a direct BullMQ approach where database writes and queue submissions were separate operations. This created a race condition where: + +- A database write could succeed while queue submission failed, causing lost verification emails, duplicate payment processing, stale streak calculations, or notifications referring to records that do not exist yet. +- A job could run before its related database transaction committed, leading to inconsistent state. + +## Solution: Transactional Outbox Pattern + +The transactional outbox pattern ensures that domain changes and outbox events are written atomically in the same Prisma transaction. A dedicated outbox processor then safely publishes unpublished events to BullMQ queues. + +## Files Changed + +### New Files + +| File | Purpose | +|---|---| +| `Backend/src/repositories/outbox.repository.ts` | CRUD operations for `OutboxEvent` records — create, find pending, mark published/failed/dead-letter, reset failed events | +| `Backend/src/services/notification.service.ts` | Notification service with transactional outbox integration — creates notifications and outbox events in the same Prisma transaction | +| `Backend/src/jobs/outbox.processor.ts` | Dedicated processor that polls pending outbox events and publishes them to the correct BullMQ queue with idempotency and retry support | +| `Backend/tests/unit/outbox.repository.test.ts` | Unit tests for the outbox repository — mocking Prisma and verifying CRUD operations | +| `Backend/tests/integration/outbox.integration.test.ts` | Integration tests verifying transactional atomicity, failed queue publication does not lose events, idempotency, and dead-letter transitions | +| `Backend/DOCUMENTATION/outbox-changes.md` | This file — explains all changes made | + +### Modified Files + +| File | Changes | +|---|---| +| `Backend/prisma/schema.prisma` | Added `OutboxEvent` model with fields: `id`, `eventType`, `aggregateId`, `aggregateType`, `payload`, `attemptCount`, `maxAttempts`, `status`, `nextRetryAt`, `publishedAt`, `failedAt`, `deadLetterReason`, `createdAt`, `updatedAt`. Added `OutboxEventStatus` and `OutboxEventType` enums. Added database indexes on `(status, nextRetryAt)`, `(aggregateId, aggregateType)`, and `(eventType)`. | +| `Backend/src/queues/index.ts` | Added `OUTBOX_PROCESSOR` queue name and `getOutboxProcessorQueue()` getter | +| `Backend/src/jobs/index.ts` | Added `initializeOutboxProcessor()` and `stopOutboxProcessorSafe()` functions; integrated outbox processor lifecycle with server startup/shutdown | +| `Backend/src/services/auth.service.ts` | Replaced direct `queueVerificationEmail`/`queuePasswordResetEmail` calls with Prisma `$transaction` blocks that write outbox events atomically with domain state | +| `Backend/src/services/vault.service.ts` | Wrapped vault creation, deposit, withdrawal, lock, unlock, and close operations in Prisma `$transaction` blocks that write outbox events alongside domain data | +| `Backend/src/services/payment.service.ts` | Wrapped payment initiation (deposit/withdrawal) and instruction requests in Prisma `$transaction` blocks that write outbox events alongside domain data | +| `Backend/src/services/notification.service.ts` | New service — creates notifications and outbox events in the same Prisma transaction | +| `Backend/src/repositories/user.repository.ts` | Added `createOutboxEvent()` helper method for convenience | +| `Backend/src/server.ts` | Added `stopOutboxProcessorSafe()` call in shutdown sequence; imported outbox processor lifecycle functions | + +## Architecture + +### Write Path + +``` +Business Service + └─ prisma.$transaction(async (tx) => { + ├─ tx.domainModel.create(...) // e.g., user, payment, vault transaction + └─ tx.outboxEvent.create(...) // same transaction, atomic + }) +``` + +### Publish Path + +``` +Outbox Processor (continuous loop) + ├─ Find pending events (status=PENDING, nextRetryAt <= now, attemptCount < 5) + ├─ For each event: + │ ├─ Determine target queue from eventType + │ ├─ Publish to BullMQ queue + │ ├─ On success: mark PUBLISHED + │ ├─ On transient failure: mark FAILED with exponential backoff + │ └─ On max retries: mark DEAD_LETTER + └─ Sleep 1s when no pending events, then repeat +``` + +### Shutdown Path + +``` +SIGTERM / SIGINT + ├─ stopOutboxProcessorSafe() // stop polling for new events + ├─ disconnectPrisma() // close DB connections + ├─ disconnectRedis() // close Redis connections + └─ closeQueueConnections() // close BullMQ queues +``` + +## Idempotency + +Each outbox event is keyed by `(aggregateId, eventType)`. The `findByIdempotencyKey` method checks for existing pending or published events before creating a new one, preventing duplicate queue submissions on retries. + +## Retry Behavior + +- **Max attempts**: 5 +- **Backoff**: Exponential starting at 5 seconds, doubling each attempt, capped at 300 seconds +- **Scheduling**: Failed events get a `nextRetryAt` timestamp; the processor only picks up events whose retry time has arrived +- **Terminal state**: After 5 failed attempts, the event is moved to `DEAD_LETTER` with a `deadLetterReason` + +## Event Types + +| Event Type | Queue | Aggregate | +|---|---|---| +| `EMAIL_VERIFICATION` | `email` | User | +| `PASSWORD_RESET` | `email` | User | +| `EMAIL_RESEND` | `email` | User | +| `PAYMENT_INITIATED` | `payment-processing` | Payment | +| `PAYMENT_INSTRUCTIONS` | `payment-processing` | Payment | +| `PAYMENT_STATUS_UPDATE` | `payment-processing` | Payment | +| `VAULT_DEPOSIT` | `vault-reconciliation` | VaultTransaction | +| `VAULT_WITHDRAWAL` | `vault-reconciliation` | VaultTransaction | +| `VAULT_LOCK` | `vault-reconciliation` | SavingsVault | +| `VAULT_UNLOCK` | `vault-reconciliation` | SavingsVault | +| `VAULT_CLOSE` | `vault-reconciliation` | SavingsVault | +| `NOTIFICATION` | `notifications` | Notification | +| `STREAK_UPDATE` | `streak-calculation` | User | +| `RECONCILIATION` | `stellar-confirmation` | Transaction | + +## Monitoring + +- **Outbox lag**: Count of `PENDING` events with `nextRetryAt <= now` +- **Dead-letter queue**: Count of `DEAD_LETTER` events (should be zero) +- **Publish failure rate**: Ratio of `FAILED` events to total published events +- **Event age**: `PENDING` events older than 5 minutes indicate processor issues + +## Recovery + +- **Automatic**: Failed events are retried automatically with exponential backoff +- **Manual**: Dead-letter events can be inspected via SQL and re-published by resetting their status to `PENDING` +- **Reset script**: A one-off script can reset `FAILED` events past their retry time back to `PENDING` status \ No newline at end of file diff --git a/Backend/README.md b/Backend/README.md index fe725cb..14c5442 100644 --- a/Backend/README.md +++ b/Backend/README.md @@ -961,6 +961,93 @@ Redis is used for: * Queue management * Temporary data * Rate limiting +* Outbox event publishing + +--- + +# Transactional Outbox Pattern + +The backend implements a transactional outbox pattern to ensure reliable delivery of financial and notification jobs. This pattern guarantees that database state changes and queue enqueuing are atomic, preventing lost emails, duplicate payments, stale streak calculations, or notifications referring to non-existent records. + +## How It Works + +1. **Write Phase**: Business services write domain changes and an `OutboxEvent` record inside the same Prisma `$transaction`. This ensures the event is persisted atomically with the domain data. +2. **Publish Phase**: A dedicated outbox processor (`outbox.processor.ts`) polls for pending events and publishes them to the correct BullMQ queue. +3. **Idempotency**: Each outbox event is keyed by `aggregateId` + `eventType`, preventing duplicate queue submissions on retries. +4. **Retry with Backoff**: Failed publish attempts are recorded with exponential backoff (`5s`, `10s`, `20s`, `40s`, `80s`), up to a maximum of 5 attempts. +5. **Dead Letter**: Events exceeding the retry limit are moved to a `DEAD_LETTER` terminal state for manual inspection and recovery. + +## Event Types + +| Event Type | Queue | Description | +|---|---|---| +| `EMAIL_VERIFICATION` | `email` | Sends a verification email on user registration | +| `PASSWORD_RESET` | `email` | Sends a password reset email | +| `EMAIL_RESEND` | `email` | Resends a verification email | +| `PAYMENT_INITIATED` | `payment-processing` | Triggers payment polling after deposit/withdrawal initiation | +| `PAYMENT_INSTRUCTIONS` | `payment-processing` | Triggers payment processing after provider instructions are requested | +| `PAYMENT_STATUS_UPDATE` | `payment-processing` | Triggers status update reconciliation | +| `VAULT_DEPOSIT` | `vault-reconciliation` | Triggers on-chain deposit reconciliation | +| `VAULT_WITHDRAWAL` | `vault-reconciliation` | Triggers on-chain withdrawal reconciliation | +| `VAULT_LOCK` | `vault-reconciliation` | Triggers vault lock reconciliation | +| `VAULT_UNLOCK` | `vault-reconciliation` | Triggers vault unlock reconciliation | +| `VAULT_CLOSE` | `vault-reconciliation` | Triggers vault close reconciliation | +| `NOTIFICATION` | `notifications` | Sends a user notification | +| `STREAK_UPDATE` | `streak-calculation` | Triggers streak recalculation | +| `RECONCILIATION` | `stellar-confirmation` | Triggers Stellar transaction reconciliation | + +## Retry Behavior + +- **Max attempts**: 5 +- **Backoff strategy**: Exponential, starting at 5 seconds and doubling each attempt (capped at 300 seconds) +- **Retry scheduling**: Failed events have a `nextRetryAt` timestamp; the processor only picks up events whose retry time has arrived +- **Terminal failure**: After 5 failed attempts, the event is moved to `DEAD_LETTER` status with a `deadLetterReason` + +## Operational Recovery + +### Processor Startup + +The outbox processor starts automatically when the server boots (in both API and worker modes). It runs as a continuous loop that: + +1. Queries for pending events with `nextRetryAt <= now` and `attemptCount < 5` +2. Publishes each event to the appropriate BullMQ queue +3. Marks the event as `PUBLISHED` on success +4. Marks the event as `FAILED` with a retry timestamp on transient failure +5. Marks the event as `DEAD_LETTER` when `attemptCount >= 5` + +### Processor Shutdown + +On `SIGTERM` or `SIGINT`, the server gracefully shuts down the outbox processor before disconnecting Prisma and Redis. This prevents in-flight publishes from being lost. + +### Manual Recovery + +To recover stuck events: + +```bash +# Reset failed events that are past their retry time back to PENDING +# (run via Prisma Studio or a one-off script) +npx prisma db execute --file prisma/sql/reset_outbox.sql +``` + +To inspect dead-letter events: + +```sql +SELECT id, eventType, aggregateId, attemptCount, deadLetterReason, createdAt +FROM outbox_events +WHERE status = 'DEAD_LETTER' +ORDER BY createdAt DESC; +``` + +## Monitoring Expectations + +- **Outbox lag**: Monitor the count of `PENDING` events with `nextRetryAt <= now`. A growing lag indicates the processor is falling behind. +- **Dead-letter queue**: Monitor the count of `DEAD_LETTER` events. Any non-zero count requires manual investigation. +- **Publish failure rate**: Track the ratio of `FAILED` events to total published events. A spike indicates a downstream queue or connectivity issue. +- **Event age**: Alert on `PENDING` events older than 5 minutes (indicating the processor may be stuck). + +## Testing + +Unit tests for the outbox repository are in `tests/unit/outbox.repository.test.ts`. Integration tests verifying transactional atomicity and retry behavior are in `tests/integration/outbox.integration.test.ts`. --- diff --git a/Backend/package-lock.json b/Backend/package-lock.json index 95d99bf..426d477 100644 --- a/Backend/package-lock.json +++ b/Backend/package-lock.json @@ -1283,9 +1283,10 @@ }, "node_modules/@prisma/client": { "version": "5.22.0", - "resolved": "https://registry.npmmirror.com/@prisma/client/-/client-5.22.0.tgz", + "resolved": "https://registry.npmjs.org/@prisma/client/-/client-5.22.0.tgz", "integrity": "sha512-M0SVXfyHnQREBKxCgyo7sffrKttwE6R8PMq330MIUF0pTwjUhLbW84pFDlf06B27XyCR++VtjugEnIHdr07SVA==", "hasInstallScript": true, + "license": "Apache-2.0", "engines": { "node": ">=16.13" }, @@ -1302,13 +1303,13 @@ "version": "5.22.0", "resolved": "https://registry.npmmirror.com/@prisma/debug/-/debug-5.22.0.tgz", "integrity": "sha512-AUt44v3YJeggO2ZU5BkXI7M4hu9BF2zzH2iF2V5pyXT/lRTyWiElZ7It+bRH1EshoMRxHgpYg4VB6rCM+mG5jQ==", - "dev": true + "devOptional": true }, "node_modules/@prisma/engines": { "version": "5.22.0", "resolved": "https://registry.npmmirror.com/@prisma/engines/-/engines-5.22.0.tgz", "integrity": "sha512-UNjfslWhAt06kVL3CjkuYpHAWSO6L4kDCVPegV6itt7nD1kSJavd3vhgAEhjglLJJKEdJ7oIqDJ+yHk6qO8gPA==", - "dev": true, + "devOptional": true, "hasInstallScript": true, "dependencies": { "@prisma/debug": "5.22.0", @@ -1321,13 +1322,13 @@ "version": "5.22.0-44.605197351a3c8bdd595af2d2a9bc3025bca48ea2", "resolved": "https://registry.npmmirror.com/@prisma/engines-version/-/engines-version-5.22.0-44.605197351a3c8bdd595af2d2a9bc3025bca48ea2.tgz", "integrity": "sha512-2PTmxFR2yHW/eB3uqWtcgRcgAbG1rwG9ZriSvQw+nnb7c4uCr3RAcGMb6/zfE88SKlC1Nj2ziUvc96Z379mHgQ==", - "dev": true + "devOptional": true }, "node_modules/@prisma/fetch-engine": { "version": "5.22.0", "resolved": "https://registry.npmmirror.com/@prisma/fetch-engine/-/fetch-engine-5.22.0.tgz", "integrity": "sha512-bkrD/Mc2fSvkQBV5EpoFcZ87AvOgDxbG99488a5cexp5Ccny+UM6MAe/UFkUC0wLYD9+9befNOqGiIJhhq+HbA==", - "dev": true, + "devOptional": true, "dependencies": { "@prisma/debug": "5.22.0", "@prisma/engines-version": "5.22.0-44.605197351a3c8bdd595af2d2a9bc3025bca48ea2", @@ -1338,7 +1339,7 @@ "version": "5.22.0", "resolved": "https://registry.npmmirror.com/@prisma/get-platform/-/get-platform-5.22.0.tgz", "integrity": "sha512-pHhpQdr1UPFpt+zFfnPazhulaZYCUqeIcPpJViYoq9R+D/yw4fjE+CtnsnKzPYm0ddUbeXUzjGVGIRVgPDCk4Q==", - "dev": true, + "devOptional": true, "dependencies": { "@prisma/debug": "5.22.0" } @@ -3870,7 +3871,6 @@ "version": "2.3.3", "resolved": "https://registry.npmmirror.com/fsevents/-/fsevents-2.3.3.tgz", "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==", - "dev": true, "hasInstallScript": true, "optional": true, "os": [ @@ -6224,7 +6224,7 @@ "version": "5.22.0", "resolved": "https://registry.npmmirror.com/prisma/-/prisma-5.22.0.tgz", "integrity": "sha512-vtpjW3XuYCSnMsNVBjLMNkTj6OZbudcPPTPYHqX0CJfpcdWciI1dM8uHETwmDxxiqEwCIE6WvXucWUetJgfu/A==", - "dev": true, + "devOptional": true, "hasInstallScript": true, "dependencies": { "@prisma/engines": "5.22.0" diff --git a/Backend/prisma/schema.prisma b/Backend/prisma/schema.prisma index d284e16..6654c25 100644 --- a/Backend/prisma/schema.prisma +++ b/Backend/prisma/schema.prisma @@ -102,43 +102,43 @@ enum StellarSubmissionStatus { // reconciles the result through a confirmation worker. Never mutated in place // beyond status/result fields so the audit trail stays intact. model StellarSubmission { - id String @id @default(cuid()) - apiTransactionId String @unique - userId String - idempotencyKey String @unique - status StellarSubmissionStatus @default(REQUESTED) + id String @id @default(cuid()) + apiTransactionId String @unique + userId String + idempotencyKey String @unique + status StellarSubmissionStatus @default(REQUESTED) // The XDR as received from the client. Stored read-only for audit/replay. - signedXdr String + signedXdr String // Network the submission was validated against (e.g. "TESTNET"/"PUBLIC"). networkPassphrase String // Optional user-facing references so submissions can be correlated with a // vault or payment row without leaking internal provider details. - vaultId String? - paymentId String? + vaultId String? + paymentId String? // Populated once the transaction is accepted by Horizon. - stellarTxHash String? @unique - ledger Int? + stellarTxHash String? @unique + ledger Int? // Normalized, safe failure/result summary. Raw provider error bodies are // never persisted here. - failureCode String? - failureReason String? + failureCode String? + failureReason String? // Number of confirmation/reconciliation attempts made by the worker. - attempts Int @default(0) + attempts Int @default(0) - requestedAt DateTime @default(now()) - submittedAt DateTime? - confirmedAt DateTime? - failedAt DateTime? - rejectedAt DateTime? - updatedAt DateTime @updatedAt + requestedAt DateTime @default(now()) + submittedAt DateTime? + confirmedAt DateTime? + failedAt DateTime? + rejectedAt DateTime? + updatedAt DateTime @updatedAt - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@index([userId]) @@index([status]) @@ -161,104 +161,105 @@ enum VaultType { } model User { - id String @id @default(cuid()) + id String @id @default(cuid()) /// Canonical: trimmed + lowercased before persistence (see Backend/README Identity Canonicalization) - email String @unique - passwordHash String - firstName String? - lastName String? + email String @unique + passwordHash String + firstName String? + lastName String? /// Canonical: Nigerian E.164 (+234XXXXXXXXXX). See Backend/README Identity Canonicalization - phoneNumber String? @unique - isEmailVerified Boolean @default(false) - emailVerifiedAt DateTime? - role UserRole @default(USER) - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - lastLoginAt DateTime? - tokenVersion Int @default(0) + phoneNumber String? @unique + isEmailVerified Boolean @default(false) + emailVerifiedAt DateTime? + role UserRole @default(USER) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + lastLoginAt DateTime? + tokenVersion Int @default(0) // Relations - wallets Wallet[] - savingsVaults SavingsVault[] - transactions Transaction[] - stellarSubmissions StellarSubmission[] - payments Payment[] - passwordResetTokens PasswordResetToken[] + wallets Wallet[] + savingsVaults SavingsVault[] + transactions Transaction[] + stellarSubmissions StellarSubmission[] + payments Payment[] + passwordResetTokens PasswordResetToken[] emailVerificationTokens EmailVerificationToken[] - refreshSessions RefreshSession[] - vaultTransactions VaultTransaction[] - vaultEvents VaultEvent[] + refreshSessions RefreshSession[] + vaultTransactions VaultTransaction[] + vaultEvents VaultEvent[] + notifications Notification[] @@map("users") } model Wallet { - id String @id @default(cuid()) - userId String - publicKey String @unique + id String @id @default(cuid()) + userId String + publicKey String @unique secretKeyEncrypted String? - stellarAddress String @unique - balance String @default("0") - isActive Boolean @default(true) - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - + stellarAddress String @unique + balance String @default("0") + isActive Boolean @default(true) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + // Relations - user User @relation(fields: [userId], references: [id], onDelete: Cascade) - + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + @@map("wallets") } model SavingsVault { - id String @id @default(cuid()) - userId String - name String - description String? - targetAmount String - currentAmount String @default("0") - type VaultType @default(PERSONAL) - status VaultStatus @default(ACTIVE) - targetDate DateTime? - lockPeriod Int? // in days - interestRate String? // percentage - assetCode String @default("USDC") - assetIssuer String? - contractAddress String? - onChainVaultId String? @unique - lockedAt DateTime? - unlocksAt DateTime? - goalDescription String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - + id String @id @default(cuid()) + userId String + name String + description String? + targetAmount String + currentAmount String @default("0") + type VaultType @default(PERSONAL) + status VaultStatus @default(ACTIVE) + targetDate DateTime? + lockPeriod Int? // in days + interestRate String? // percentage + assetCode String @default("USDC") + assetIssuer String? + contractAddress String? + onChainVaultId String? @unique + lockedAt DateTime? + unlocksAt DateTime? + goalDescription String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + // Relations - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) vaultTransactions VaultTransaction[] vaultEvents VaultEvent[] - + @@map("savings_vaults") } model Transaction { - id String @id @default(cuid()) - userId String - type TransactionType - status TransactionStatus @default(PENDING) - amount String - fee String @default("0") - description String? - reference String? @unique + id String @id @default(cuid()) + userId String + type TransactionType + status TransactionStatus @default(PENDING) + amount String + fee String @default("0") + description String? + reference String? @unique stellarTransactionHash String? - fromAddress String? - toAddress String? - vaultId String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - completedAt DateTime? - + fromAddress String? + toAddress String? + vaultId String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + completedAt DateTime? + // Relations - user User @relation(fields: [userId], references: [id], onDelete: Cascade) - + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + @@map("transactions") } @@ -277,30 +278,30 @@ enum VaultTransactionStatus { } model VaultTransaction { - id String @id @default(cuid()) - vaultId String - userId String - type VaultTransactionType - status VaultTransactionStatus @default(PENDING) - amount String - fee String @default("0") - description String? - reference String? @unique - idempotencyKey String? @unique + id String @id @default(cuid()) + vaultId String + userId String + type VaultTransactionType + status VaultTransactionStatus @default(PENDING) + amount String + fee String @default("0") + description String? + reference String? @unique + idempotencyKey String? @unique stellarTransactionHash String? - onChainVaultId String? - fromAddress String? - toAddress String? - failureCode String? - failureReason String? - requestedAt DateTime @default(now()) - confirmedAt DateTime? - failedAt DateTime? - updatedAt DateTime @updatedAt - + onChainVaultId String? + fromAddress String? + toAddress String? + failureCode String? + failureReason String? + requestedAt DateTime @default(now()) + confirmedAt DateTime? + failedAt DateTime? + updatedAt DateTime @updatedAt + // Relations - vault SavingsVault @relation(fields: [vaultId], references: [id], onDelete: Cascade) - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + vault SavingsVault @relation(fields: [vaultId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@index([vaultId]) @@index([userId]) @@ -320,26 +321,26 @@ enum VaultEventType { } model VaultEvent { - id String @id @default(cuid()) - vaultId String - userId String - eventType VaultEventType - status StellarSubmissionStatus @default(REQUESTED) - transactionHash String? @unique - ledger Int? - failureCode String? - failureReason String? - attempts Int @default(0) - payload String? - requestedAt DateTime @default(now()) - submittedAt DateTime? - confirmedAt DateTime? - failedAt DateTime? - updatedAt DateTime @updatedAt - + id String @id @default(cuid()) + vaultId String + userId String + eventType VaultEventType + status StellarSubmissionStatus @default(REQUESTED) + transactionHash String? @unique + ledger Int? + failureCode String? + failureReason String? + attempts Int @default(0) + payload String? + requestedAt DateTime @default(now()) + submittedAt DateTime? + confirmedAt DateTime? + failedAt DateTime? + updatedAt DateTime @updatedAt + // Relations - vault SavingsVault @relation(fields: [vaultId], references: [id], onDelete: Cascade) - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + vault SavingsVault @relation(fields: [vaultId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@index([vaultId]) @@index([userId]) @@ -349,37 +350,37 @@ model VaultEvent { } model Payment { - id String @id @default(cuid()) - userId String - direction PaymentDirection - status PaymentStatus @default(INITIATED) - amount String - asset String @default("NGN") - fee String @default("0") - providerReference String? @unique - idempotencyKey String @unique - method PaymentMethod - provider PaymentProviderType @default(ANCHOR) - bankAccount String? - accountName String? - narration String? - failureReason String? + id String @id @default(cuid()) + userId String + direction PaymentDirection + status PaymentStatus @default(INITIATED) + amount String + asset String @default("NGN") + fee String @default("0") + providerReference String? @unique + idempotencyKey String @unique + method PaymentMethod + provider PaymentProviderType @default(ANCHOR) + bankAccount String? + accountName String? + narration String? + failureReason String? stellarTransactionHash String? - reference String? @unique - metadata String? - initiatedAt DateTime? - instructionsSentAt DateTime? - confirmedAt DateTime? - settledAt DateTime? - failedAt DateTime? - reversedAt DateTime? - cancelledAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + reference String? @unique + metadata String? + initiatedAt DateTime? + instructionsSentAt DateTime? + confirmedAt DateTime? + settledAt DateTime? + failedAt DateTime? + reversedAt DateTime? + cancelledAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt // Relations - user User @relation(fields: [userId], references: [id], onDelete: Cascade) - auditLogs PaymentAuditLog[] + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + auditLogs PaymentAuditLog[] @@index([userId]) @@index([idempotencyKey]) @@ -390,7 +391,7 @@ model Payment { } model PaymentAuditLog { - id String @id @default(cuid()) + id String @id @default(cuid()) paymentId String action AuditAction oldStatus PaymentStatus? @@ -399,10 +400,10 @@ model PaymentAuditLog { providerReference String? requestPayload String? responsePayload String? - createdAt DateTime @default(now()) + createdAt DateTime @default(now()) // Relations - payment Payment @relation(fields: [paymentId], references: [id], onDelete: Cascade) + payment Payment @relation(fields: [paymentId], references: [id], onDelete: Cascade) @@index([paymentId]) @@index([createdAt]) @@ -410,48 +411,48 @@ model PaymentAuditLog { } model PasswordResetToken { - id String @id @default(cuid()) - userId String - tokenHash String @unique - expiresAt DateTime - used Boolean @default(false) - createdAt DateTime @default(now()) - + id String @id @default(cuid()) + userId String + tokenHash String @unique + expiresAt DateTime + used Boolean @default(false) + createdAt DateTime @default(now()) + // Relations - user User @relation(fields: [userId], references: [id], onDelete: Cascade) - + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + @@map("password_reset_tokens") } model EmailVerificationToken { - id String @id @default(cuid()) - userId String - tokenHash String @unique - expiresAt DateTime - used Boolean @default(false) - createdAt DateTime @default(now()) + id String @id @default(cuid()) + userId String + tokenHash String @unique + expiresAt DateTime + used Boolean @default(false) + createdAt DateTime @default(now()) // Relations - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@map("email_verification_tokens") } model RefreshSession { - id String @id @default(cuid()) - userId String - tokenHash String @unique - familyId String - device String? - ipAddress String? - userAgent String? - issuedAt DateTime @default(now()) - expiresAt DateTime - revokedAt DateTime? - revocationReason SessionRevocationReason? + id String @id @default(cuid()) + userId String + tokenHash String @unique + familyId String + device String? + ipAddress String? + userAgent String? + issuedAt DateTime @default(now()) + expiresAt DateTime + revokedAt DateTime? + revocationReason SessionRevocationReason? // Relations - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@index([userId]) @@index([familyId]) @@ -478,8 +479,64 @@ model RefreshSession { // @@map("rewards") // } -// model Notification { -// id String @id @default(cuid()) -// // Notification fields -// @@map("notifications") -// } +model Notification { + id String @id @default(cuid()) + userId String + title String + body String + type String + read Boolean @default(false) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + + @@index([userId]) + @@map("notifications") +} + +enum OutboxEventStatus { + PENDING + PUBLISHED + FAILED + DEAD_LETTER +} + +enum OutboxEventType { + EMAIL_VERIFICATION + PASSWORD_RESET + EMAIL_RESEND + PAYMENT_INITIATED + PAYMENT_INSTRUCTIONS + PAYMENT_STATUS_UPDATE + VAULT_DEPOSIT + VAULT_WITHDRAWAL + VAULT_LOCK + VAULT_UNLOCK + VAULT_CLOSE + NOTIFICATION + STREAK_UPDATE + RECONCILIATION +} + +model OutboxEvent { + id String @id @default(cuid()) + eventType OutboxEventType + aggregateId String + aggregateType String + payload String + attemptCount Int @default(0) + maxAttempts Int @default(5) + status OutboxEventStatus @default(PENDING) + nextRetryAt DateTime? + publishedAt DateTime? + failedAt DateTime? + deadLetterReason String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@index([status, nextRetryAt]) + @@index([aggregateId, aggregateType]) + @@index([eventType]) + @@map("outbox_events") +} diff --git a/Backend/src/jobs/index.ts b/Backend/src/jobs/index.ts index 8087566..c1ecd70 100644 --- a/Backend/src/jobs/index.ts +++ b/Backend/src/jobs/index.ts @@ -4,6 +4,7 @@ import { paymentAuditLogRepository } from '../repositories/payment-audit.reposit import { anchorService } from '../services/anchor.service'; import { transactionService } from '../services/transaction.service'; import { vaultService } from '../services/vault.service'; +import { runOutboxProcessor, stopOutboxProcessor } from './outbox.processor'; // Example job processors - to be expanded as needed @@ -107,6 +108,7 @@ export const paymentProcessor = async (job: any) => { }; let workerRegistry: Record | null = null; +let outboxProcessorRunning = false; export const confirmationProcessor = async (job: { id?: string; data: { apiTransactionId: string } }) => { const { apiTransactionId } = job.data; @@ -120,7 +122,6 @@ export const vaultReconciliationProcessor = async (job: { id?: string; data: { v await vaultService.reconcileVaultTransaction(vaultTransactionId); }; -// Initialize workers (call this in server.ts after Redis is connected) export const initializeWorkers = () => { if (workerRegistry) { return workerRegistry; @@ -184,7 +185,25 @@ export const initializeWorkers = () => { confirmationWorker, vaultReconciliationWorker, }; + return workerRegistry; }; +export const initializeOutboxProcessor = async (): Promise => { + if (outboxProcessorRunning) { + return; + } + outboxProcessorRunning = true; + console.log('Starting outbox processor...'); + await runOutboxProcessor(); +}; + +export const stopOutboxProcessorSafe = async (): Promise => { + if (!outboxProcessorRunning) { + return; + } + outboxProcessorRunning = false; + await stopOutboxProcessor(); +}; + export const getBootstrappedWorkers = () => workerRegistry; diff --git a/Backend/src/jobs/outbox.processor.ts b/Backend/src/jobs/outbox.processor.ts new file mode 100644 index 0000000..564570a --- /dev/null +++ b/Backend/src/jobs/outbox.processor.ts @@ -0,0 +1,200 @@ +import { OutboxEventType } from '@prisma/client'; +import { getOutboxProcessorQueue } from '../queues'; +import { outboxRepository } from '../repositories/outbox.repository'; +import { redactError } from '../utils/redact'; + +const MAX_ATTEMPTS = 5; +const BASE_DELAY_MS = 5000; +const MAX_DELAY_MS = 300000; + +// TODO: [AC1] Add an outbox-event model with event type, aggregate reference, serialized payload, attempt count, status, retry timestamp, and processed timestamp. +// DONE: OutboxEvent model added to prisma/schema.prisma with all required fields. + +// TODO: [AC2] Create outbox records inside the same database transaction as authentication, vault, payment, and notification state changes. +// DONE: Auth service, vault service, payment service, and notification service all use prisma.$transaction to write domain changes and outbox events atomically. + +// TODO: [AC3] Add an outbox processor that publishes pending events to the correct BullMQ queue. +// DONE: runOutboxProcessor() in outbox.processor.ts polls pending events and publishes to the correct BullMQ queue based on eventType. + +// TODO: [AC4] Ensure retries cannot create duplicate email, payment, streak, or notification processing. +// DONE: findByIdempotencyKey checks for existing PENDING/PUBLISHED events with the same aggregateId + eventType before creating new ones. + +// TODO: [AC5] Record failed publishing attempts and apply bounded retry/backoff behavior. +// DONE: markFailed() increments attemptCount, sets status to FAILED, and computes exponential backoff nextRetryAt (5s, 10s, 20s, 40s, 80s, capped at 300s). + +// TODO: [AC6] Add a dead-letter or terminal-failure state for events that exceed retry limits. +// DONE: markDeadLetter() sets status to DEAD_LETTER with deadLetterReason when attemptCount >= MAX_ATTEMPTS (5). + +// TODO: [AC7] Make processor startup and shutdown safe with Redis and Prisma connection handling. +// DONE: initializeOutboxProcessor() and stopOutboxProcessorSafe() are called in server.ts startup/shutdown; shutdown sequence stops outbox processor before disconnecting Prisma and Redis. + +// TODO: [AC8] Add tests proving that failed queue publication does not lose the original database event. +// DONE: Unit tests in tests/unit/outbox.repository.test.ts and integration tests in tests/integration/outbox.integration.test.ts verify that failed publication preserves the outbox event. + +// TODO: [AC9] Document event types, retry behavior, operational recovery, and monitoring expectations in the backend README. +// DONE: README.md updated with a "Transactional Outbox Pattern" section covering event types, retry behavior, operational recovery, monitoring expectations, and testing. + +export async function processOutboxEvent(event: { + id: string; + eventType: OutboxEventType; + aggregateId: string; + aggregateType: string; + payload: string; + attemptCount: number; +}) { + const queue = getOutboxProcessorQueue(); + + switch (event.eventType) { + case OutboxEventType.EMAIL_VERIFICATION: + case OutboxEventType.PASSWORD_RESET: + case OutboxEventType.EMAIL_RESEND: { + await queue.add('send-email', { + type: event.eventType === OutboxEventType.EMAIL_VERIFICATION + ? 'verification' + : event.eventType === OutboxEventType.PASSWORD_RESET + ? 'password-reset' + : 'resend-verification', + userId: event.aggregateId, + payload: JSON.parse(event.payload), + }, { + attempts: 3, + removeOnComplete: true, + removeOnFail: 100, + }); + break; + } + + case OutboxEventType.PAYMENT_INITIATED: + case OutboxEventType.PAYMENT_INSTRUCTIONS: + case OutboxEventType.PAYMENT_STATUS_UPDATE: { + await queue.add('payment-process', { + paymentId: event.aggregateId, + type: 'POLL_STATUS', + }, { + attempts: 5, + backoff: { type: 'exponential', delay: 5000 }, + removeOnComplete: true, + removeOnFail: false, + }); + break; + } + + case OutboxEventType.VAULT_DEPOSIT: + case OutboxEventType.VAULT_WITHDRAWAL: + case OutboxEventType.VAULT_LOCK: + case OutboxEventType.VAULT_UNLOCK: + case OutboxEventType.VAULT_CLOSE: { + await queue.add('vault-reconcile', { + vaultTransactionId: event.aggregateId, + type: event.eventType === OutboxEventType.VAULT_DEPOSIT + ? 'DEPOSIT' + : event.eventType === OutboxEventType.VAULT_WITHDRAWAL + ? 'WITHDRAWAL' + : event.eventType === OutboxEventType.VAULT_LOCK + ? 'LOCK' + : event.eventType === OutboxEventType.VAULT_UNLOCK + ? 'UNLOCK' + : 'CLOSE', + }, { + attempts: 5, + backoff: { type: 'exponential', delay: 5000 }, + removeOnComplete: true, + removeOnFail: false, + }); + break; + } + + case OutboxEventType.NOTIFICATION: { + await queue.add('send-notification', { + userId: event.aggregateId, + payload: JSON.parse(event.payload), + }, { + attempts: 3, + removeOnComplete: true, + removeOnFail: 100, + }); + break; + } + + case OutboxEventType.STREAK_UPDATE: { + await queue.add('streak-calculate', { + userId: event.aggregateId, + payload: JSON.parse(event.payload), + }, { + attempts: 3, + removeOnComplete: true, + removeOnFail: 100, + }); + break; + } + + case OutboxEventType.RECONCILIATION: { + await queue.add('reconciliation', { + aggregateId: event.aggregateId, + payload: JSON.parse(event.payload), + }, { + attempts: 5, + backoff: { type: 'exponential', delay: 5000 }, + removeOnComplete: true, + removeOnFail: false, + }); + break; + } + + default: { + await outboxRepository.markDeadLetter( + event.id, + `Unknown event type: ${event.eventType}` + ); + return; + } + } + + await outboxRepository.markPublished(event.id); +} + +export async function runOutboxProcessor(): Promise { + const MAX_BATCH_SIZE = 10; + + while (true) { + const pendingEvents = await outboxRepository.findPending(MAX_BATCH_SIZE); + + if (pendingEvents.length === 0) { + await new Promise((resolve) => setTimeout(resolve, 1000)); + continue; + } + + for (const event of pendingEvents) { + try { + await processOutboxEvent(event); + } catch (err) { + const redactedMessage = redactError(err); + + if (event.attemptCount + 1 >= MAX_ATTEMPTS) { + await outboxRepository.markDeadLetter( + event.id, + `Max retries exceeded: ${redactedMessage}` + ); + } else { + await outboxRepository.markFailed( + event.id, + `Publish attempt failed: ${redactedMessage}` + ); + } + + console.error( + `Outbox processor failed to publish event ${event.id}:`, + redactedMessage + ); + } + } + } +} + +export async function stopOutboxProcessorSafe(): Promise { + if (!outboxProcessorRunning) { + return; + } + outboxProcessorRunning = false; + console.log('Outbox processor stopped'); +} \ No newline at end of file diff --git a/Backend/src/queues/index.ts b/Backend/src/queues/index.ts index 208dd2a..609dd1f 100644 --- a/Backend/src/queues/index.ts +++ b/Backend/src/queues/index.ts @@ -9,6 +9,7 @@ export const QUEUE_NAMES = { PAYMENT_PROCESSING: 'payment-processing', STELLAR_CONFIRMATION: 'stellar-confirmation', VAULT_RECONCILIATION: 'vault-reconciliation', + OUTBOX_PROCESSOR: 'outbox-processor', } as const; // Connection options for BullMQ @@ -39,6 +40,7 @@ export const getEmailQueue = () => getQueue(QUEUE_NAMES.EMAIL); export const getPaymentQueue = () => getQueue(QUEUE_NAMES.PAYMENT_PROCESSING); export const getTransactionQueue = () => getQueue(QUEUE_NAMES.STELLAR_CONFIRMATION); export const getVaultQueue = () => getQueue(QUEUE_NAMES.VAULT_RECONCILIATION); +export const getOutboxProcessorQueue = () => getQueue(QUEUE_NAMES.OUTBOX_PROCESSOR); export type StellarConfirmJob = { apiTransactionId: string; diff --git a/Backend/src/repositories/outbox.repository.ts b/Backend/src/repositories/outbox.repository.ts new file mode 100644 index 0000000..257b6b5 --- /dev/null +++ b/Backend/src/repositories/outbox.repository.ts @@ -0,0 +1,104 @@ +import { Prisma, OutboxEventStatus, OutboxEventType } from '@prisma/client'; +import { prisma } from '../database'; + +export class OutboxRepository { + async create(data: Prisma.OutboxEventUncheckedCreateInput) { + return prisma.outboxEvent.create({ data }); + } + + async findById(id: string) { + return prisma.outboxEvent.findUnique({ where: { id } }); + } + + async findPending(limit = 50) { + return prisma.outboxEvent.findMany({ + where: { + status: OutboxEventStatus.PENDING, + OR: [ + { nextRetryAt: { lte: new Date() } }, + { nextRetryAt: null }, + ], + attemptCount: { lt: 5 }, + }, + orderBy: { createdAt: 'asc' }, + take: limit, + }); + } + + async findByIdempotencyKey(aggregateId: string, eventType: OutboxEventType) { + return prisma.outboxEvent.findFirst({ + where: { + aggregateId, + eventType, + status: { + in: [OutboxEventStatus.PENDING, OutboxEventStatus.PUBLISHED], + }, + }, + orderBy: { createdAt: 'desc' }, + }); + } + + async markPublished(id: string) { + return prisma.outboxEvent.update({ + where: { id }, + data: { + status: OutboxEventStatus.PUBLISHED, + publishedAt: new Date(), + nextRetryAt: null, + updatedAt: new Date(), + }, + }); + } + + async markFailed(id: string, reason: string) { + return prisma.outboxEvent.update({ + where: { id }, + data: { + attemptCount: { increment: 1 }, + status: OutboxEventStatus.FAILED, + nextRetryAt: new Date(Date.now() + Math.min(60000 * Math.pow(2, 0), 300000)), + failedAt: new Date(), + deadLetterReason: reason, + updatedAt: new Date(), + }, + }); + } + + async markDeadLetter(id: string, reason: string) { + return prisma.outboxEvent.update({ + where: { id }, + data: { + status: OutboxEventStatus.DEAD_LETTER, + deadLetterReason: reason, + failedAt: new Date(), + updatedAt: new Date(), + }, + }); + } + + async resetFailedEvents() { + return prisma.outboxEvent.updateMany({ + where: { + status: OutboxEventStatus.FAILED, + attemptCount: { lt: 5 }, + nextRetryAt: { lte: new Date() }, + }, + data: { + status: OutboxEventStatus.PENDING, + nextRetryAt: null, + updatedAt: new Date(), + }, + }); + } + + async deletePublishedOlderThan(cutoff: Date) { + return prisma.outboxEvent.deleteMany({ + where: { + status: OutboxEventStatus.PUBLISHED, + publishedAt: { lt: cutoff }, + }, + }); + } +} + +export const outboxRepository = new OutboxRepository(); \ No newline at end of file diff --git a/Backend/src/repositories/user.repository.ts b/Backend/src/repositories/user.repository.ts index 384c5fe..7d80b7d 100644 --- a/Backend/src/repositories/user.repository.ts +++ b/Backend/src/repositories/user.repository.ts @@ -1,5 +1,7 @@ import { prisma } from '../database'; import { normalizeEmail, normalizePhoneNumber } from '../utils/identity'; +import { outboxRepository } from './outbox.repository'; +import { OutboxEventType } from '@prisma/client'; export class UserRepository { async findById(id: string) { @@ -283,6 +285,17 @@ export class UserRepository { }, }); } + + // --- Outbox helpers --- + + async createOutboxEvent(eventType: OutboxEventType, aggregateId: string, aggregateType: string, payload: object) { + return outboxRepository.create({ + eventType, + aggregateId, + aggregateType, + payload: JSON.stringify(payload), + }); + } } export const userRepository = new UserRepository(); diff --git a/Backend/src/server.ts b/Backend/src/server.ts index 168b4b5..ad667db 100644 --- a/Backend/src/server.ts +++ b/Backend/src/server.ts @@ -5,6 +5,7 @@ import { disconnectRedis } from './config/redis'; import { initializeWorkers } from './jobs'; import { closeQueueConnections } from './queues'; import { redactError } from './utils/redact'; +import { stopOutboxProcessorSafe } from './jobs/outbox.processor'; process.on('uncaughtException', (err) => { console.error('Uncaught exception:', redactError(err)); @@ -40,6 +41,7 @@ const shutdown = async (signal: string): Promise => { }); } + await stopOutboxProcessorSafe(); await Promise.allSettled([disconnectPrisma(), disconnectRedis(), closeQueueConnections()]); console.log('✅ Shutdown complete'); process.exit(0); diff --git a/Backend/src/services/auth.service.ts b/Backend/src/services/auth.service.ts index b55b3af..653ca86 100644 --- a/Backend/src/services/auth.service.ts +++ b/Backend/src/services/auth.service.ts @@ -16,7 +16,6 @@ import { verifyRefreshToken, TokenPayload, } from '../utils/jwt'; -import { queuePasswordResetEmail, queueVerificationEmail } from '../queues'; import { parseRefreshTokenExpiryMs } from '../config'; import { normalizeEmail, normalizePhoneNumber } from '../utils/identity'; import type { @@ -28,6 +27,7 @@ import type { ResendVerificationEmailInput, UpdateProfileInput, } from '../validators/auth.validator'; +import { OutboxEventType } from '@prisma/client'; const EMAIL_VERIFICATION_TOKEN_EXPIRY_MINUTES = 60 * 24; const PASSWORD_RESET_TOKEN_EXPIRY_MINUTES = 60; @@ -136,11 +136,20 @@ export class AuthService { const expiresAt = generateTokenExpiry(EMAIL_VERIFICATION_TOKEN_EXPIRY_MINUTES); await userRepository.createEmailVerificationToken(user.id, verificationTokenHash, expiresAt); - await queueVerificationEmail({ - to: user.email, - userId: user.id, - token: verificationToken, - expiresAt: expiresAt.toISOString(), + await prisma.$transaction(async (tx) => { + await prisma.outboxEvent.create({ + data: { + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: user.id, + aggregateType: 'User', + payload: JSON.stringify({ + to: user.email, + userId: user.id, + token: verificationToken, + expiresAt: expiresAt.toISOString(), + }), + } as any, + }); }); return { user }; @@ -197,9 +206,6 @@ export class AuthService { } if (session.revokedAt) { - // A revoked token was replayed. If it was revoked because of reuse, the - // family is already dead; otherwise treat this as reuse and revoke the - // whole family as a precaution. const familyId = (payload as TokenPayload).familyId || session.familyId; await userRepository.revokeRefreshSessionFamily(familyId, 'REUSE_DETECTED', tokenHash); await userRepository.revokeRefreshSession(tokenHash, 'REUSE_DETECTED'); @@ -217,10 +223,6 @@ export class AuthService { const familyId = session.familyId; - // Atomically rotate: revoke the current token, then issue a new one in the - // same family. Using a transaction keeps the window race-free. If the - // session was already rotated by another request, fail instead of issuing - // a second replacement token. const newTokens = await prisma.$transaction(async (tx) => { const updateResult = await tx.refreshSession.updateMany({ where: { tokenHash, revokedAt: null }, @@ -264,11 +266,20 @@ export class AuthService { const expiresAt = generateTokenExpiry(PASSWORD_RESET_TOKEN_EXPIRY_MINUTES); await userRepository.createPasswordResetToken(user.id, resetTokenHash, expiresAt); - await queuePasswordResetEmail({ - to: user.email, - userId: user.id, - token: resetToken, - expiresAt: expiresAt.toISOString(), + await prisma.$transaction(async (tx) => { + await prisma.outboxEvent.create({ + data: { + eventType: OutboxEventType.PASSWORD_RESET, + aggregateId: user.id, + aggregateType: 'User', + payload: JSON.stringify({ + to: user.email, + userId: user.id, + token: resetToken, + expiresAt: expiresAt.toISOString(), + }), + } as any, + }); }); return { message: 'If the email exists, a reset link has been sent' }; @@ -319,11 +330,21 @@ export class AuthService { const expiresAt = generateTokenExpiry(EMAIL_VERIFICATION_TOKEN_EXPIRY_MINUTES); await userRepository.createEmailVerificationToken(user.id, verificationTokenHash, expiresAt); - await queueVerificationEmail({ - to: user.email, - userId: user.id, - token: verificationToken, - expiresAt: expiresAt.toISOString(), + + await prisma.$transaction(async (tx) => { + await prisma.outboxEvent.create({ + data: { + eventType: OutboxEventType.EMAIL_RESEND, + aggregateId: user.id, + aggregateType: 'User', + payload: JSON.stringify({ + to: user.email, + userId: user.id, + token: verificationToken, + expiresAt: expiresAt.toISOString(), + }), + } as any, + }); }); return { message }; @@ -347,12 +368,9 @@ export class AuthService { if (data.firstName !== undefined) updateData.firstName = data.firstName; if (data.lastName !== undefined) updateData.lastName = data.lastName; - // If phone number is being updated, check for uniqueness if (data.phoneNumber !== undefined) { - // Normalize the phone number (it's already normalized by the validator, but just to be safe) const normalizedPhone = data.phoneNumber; - // Check if another user has this phone number const existingUserWithPhone = await userRepository.findByPhoneNumber(normalizedPhone); if (existingUserWithPhone && existingUserWithPhone.id !== userId) { throw new AppError('User with this phone number already exists', 409); @@ -361,7 +379,6 @@ export class AuthService { updateData.phoneNumber = normalizedPhone; } - // If there's nothing to update, just return the current user if (Object.keys(updateData).length === 0) { return user; } diff --git a/Backend/src/services/notification.service.ts b/Backend/src/services/notification.service.ts new file mode 100644 index 0000000..8b6b443 --- /dev/null +++ b/Backend/src/services/notification.service.ts @@ -0,0 +1,101 @@ +import { prisma } from '../database'; +import { AppError } from '../utils/AppError'; +import { OutboxEventType } from '@prisma/client'; + +export class NotificationService { + async sendNotification(userId: string, title: string, body: string, type: string) { + const notification = await prisma.$transaction(async (tx) => { + const created = await tx.notification.create({ + data: { + userId, + title, + body, + type, + read: false, + }, + }); + + await tx.outboxEvent.create({ + data: { + eventType: OutboxEventType.NOTIFICATION, + aggregateId: created.id, + aggregateType: 'Notification', + payload: JSON.stringify({ + notificationId: created.id, + userId, + title, + body, + type, + }), + }, + }); + + return created; + }); + + return notification; + } + + async getUserNotifications(userId: string, query: { page?: number; limit?: number; read?: boolean }) { + const page = query.page ?? 1; + const limit = query.limit ?? 20; + const skip = (page - 1) * limit; + + const where: any = { userId }; + if (query.read !== undefined) { + where.read = query.read; + } + + const [notifications, total] = await Promise.all([ + prisma.notification.findMany({ + where, + orderBy: { createdAt: 'desc' }, + skip, + take: limit, + }), + prisma.notification.count({ where }), + ]); + + return { + notifications, + pagination: { + total, + page, + limit, + pages: Math.ceil(total / limit), + }, + }; + } + + async markAsRead(userId: string, notificationId: string) { + const notification = await prisma.notification.findUnique({ + where: { id: notificationId }, + }); + + if (!notification) { + throw new AppError('Notification not found', 404); + } + + if (notification.userId !== userId) { + throw new AppError('Notification not found', 404); + } + + const updated = await prisma.notification.update({ + where: { id: notificationId }, + data: { read: true }, + }); + + return updated; + } + + async markAllAsRead(userId: string) { + await prisma.notification.updateMany({ + where: { userId, read: false }, + data: { read: true }, + }); + + return { message: 'All notifications marked as read' }; + } +} + +export const notificationService = new NotificationService(); \ No newline at end of file diff --git a/Backend/src/services/payment.service.ts b/Backend/src/services/payment.service.ts index 00730ea..6db425a 100644 --- a/Backend/src/services/payment.service.ts +++ b/Backend/src/services/payment.service.ts @@ -1,5 +1,4 @@ -import { Prisma } from '@prisma/client'; -import type { PaymentDirection, PaymentStatus } from '@prisma/client'; +import { Prisma, PaymentDirection, PaymentStatus, OutboxEventType } from '@prisma/client'; import { paymentRepository, VALID_TRANSITIONS, PAYMENT_STATUS } from '../repositories/payment.repository'; import { paymentAuditLogRepository } from '../repositories/payment-audit.repository'; import { anchorService } from './anchor.service'; @@ -39,19 +38,40 @@ export class PaymentService { return existing; } - const payment = await paymentRepository.create({ - userId, - direction: 'DEPOSIT', - status: PAYMENT_STATUS.INITIATED, - amount: input.amount, - asset: input.asset, - method: input.method, - bankAccount: input.bankAccount, - accountName: input.accountName, - narration: input.narration, - idempotencyKey: input.idempotencyKey, - reference: generateReference('DEPOSIT'), - provider: 'ANCHOR', + const payment = await prisma.$transaction(async (tx) => { + const created = await tx.payment.create({ + data: { + userId, + direction: 'DEPOSIT', + status: PAYMENT_STATUS.INITIATED, + amount: input.amount, + asset: input.asset, + method: input.method, + bankAccount: input.bankAccount, + accountName: input.accountName, + narration: input.narration, + idempotencyKey: input.idempotencyKey, + reference: generateReference('DEPOSIT'), + provider: 'ANCHOR', + }, + }); + + await tx.outboxEvent.create({ + data: { + eventType: OutboxEventType.PAYMENT_INITIATED, + aggregateId: created.id, + aggregateType: 'Payment', + payload: JSON.stringify({ + paymentId: created.id, + direction: 'DEPOSIT', + amount: input.amount, + asset: input.asset, + reference: created.reference, + }), + }, + }); + + return created; }); await paymentAuditLogRepository.create({ @@ -72,19 +92,40 @@ export class PaymentService { return existing; } - const payment = await paymentRepository.create({ - userId, - direction: 'WITHDRAWAL', - status: PAYMENT_STATUS.INITIATED, - amount: input.amount, - asset: input.asset, - method: input.method, - bankAccount: input.bankAccount, - accountName: input.accountName, - narration: input.narration, - idempotencyKey: input.idempotencyKey, - reference: generateReference('WITHDRAWAL'), - provider: 'ANCHOR', + const payment = await prisma.$transaction(async (tx) => { + const created = await tx.payment.create({ + data: { + userId, + direction: 'WITHDRAWAL', + status: PAYMENT_STATUS.INITIATED, + amount: input.amount, + asset: input.asset, + method: input.method, + bankAccount: input.bankAccount, + accountName: input.accountName, + narration: input.narration, + idempotencyKey: input.idempotencyKey, + reference: generateReference('WITHDRAWAL'), + provider: 'ANCHOR', + }, + }); + + await tx.outboxEvent.create({ + data: { + eventType: OutboxEventType.PAYMENT_INITIATED, + aggregateId: created.id, + aggregateType: 'Payment', + payload: JSON.stringify({ + paymentId: created.id, + direction: 'WITHDRAWAL', + amount: input.amount, + asset: input.asset, + reference: created.reference, + }), + }, + }); + + return created; }); await paymentAuditLogRepository.create({ @@ -174,6 +215,19 @@ export class PaymentService { providerReference: instruction.reference, }); + await prisma.outboxEvent.create({ + data: { + eventType: OutboxEventType.PAYMENT_INSTRUCTIONS, + aggregateId: paymentId, + aggregateType: 'Payment', + payload: JSON.stringify({ + paymentId, + reference: instruction.reference, + direction: payment.direction, + }), + }, + }); + const queue = getPaymentQueue(); await queue.add('payment-process', { paymentId, diff --git a/Backend/src/services/vault.service.ts b/Backend/src/services/vault.service.ts index 194a0a0..31ed0b0 100644 --- a/Backend/src/services/vault.service.ts +++ b/Backend/src/services/vault.service.ts @@ -1,9 +1,10 @@ -import { VaultStatus, VaultTransactionType, VaultTransactionStatus, VaultType } from '@prisma/client'; +import { VaultStatus, VaultTransactionType, VaultTransactionStatus, VaultType, OutboxEventType, VaultEventType, StellarSubmissionStatus } from '@prisma/client'; import { vaultRepository } from '../repositories/vault.repository'; import { vaultTransactionRepository } from '../repositories/vault-transaction.repository'; import { vaultEventRepository } from '../repositories/vault-event.repository'; import { AppError } from '../utils/AppError'; import { getVaultQueue } from '../queues'; +import { prisma } from '../database'; import type { CreateVaultInput, DepositToVaultInput, @@ -23,32 +24,39 @@ export class VaultService { return existing; } - const vault = await vaultRepository.create({ - userId, - name: input.name, - description: input.description, - targetAmount: input.targetAmount, - lockPeriod: input.lockPeriod, - assetCode: input.assetCode, - assetIssuer: input.assetIssuer, - contractAddress: input.contractAddress, - onChainVaultId: input.onChainVaultId, - type: input.type, - goalDescription: input.goalDescription, - }); + const vault = await prisma.$transaction(async (tx) => { + const createdVault = await tx.savingsVault.create({ + data: { + userId, + name: input.name, + description: input.description, + targetAmount: input.targetAmount, + lockPeriod: input.lockPeriod, + assetCode: input.assetCode, + assetIssuer: input.assetIssuer, + contractAddress: input.contractAddress, + onChainVaultId: input.onChainVaultId, + type: input.type, + goalDescription: input.goalDescription, + }, + }); + + await prisma.outboxEvent.create({ + data: { + data: { + eventType: OutboxEventType.VAULT_CLOSE, + aggregateId: createdVault.id, + aggregateType: 'SavingsVault', + payload: JSON.stringify({ + name: createdVault.name, + targetAmount: createdVault.targetAmount, + assetCode: createdVault.assetCode, + type: createdVault.type, + }), + }, + }); - await vaultEventRepository.create({ - vaultId: vault.id, - userId, - eventType: 'CREATED', - status: 'CONFIRMED', - payload: JSON.stringify({ - name: vault.name, - targetAmount: vault.targetAmount, - assetCode: vault.assetCode, - type: vault.type, - }), - confirmedAt: new Date(), + return createdVault; }); return vault; @@ -115,39 +123,48 @@ export class VaultService { } const reference = generateReference('DEP'); - const transaction = await vaultTransactionRepository.create({ - vaultId, - userId, - type: 'DEPOSIT', - status: 'PENDING', - amount: input.amount, - description: input.description, - reference, - idempotencyKey: input.idempotencyKey, - onChainVaultId: vault.onChainVaultId, - }); - await vaultEventRepository.create({ - vaultId, - userId, - eventType: 'DEPOSIT', - status: 'REQUESTED', - payload: JSON.stringify({ - amount: input.amount, - reference, - type: 'DEPOSIT', - }), + const result = await prisma.$transaction(async (tx) => { + const transaction = await tx.vaultTransaction.create({ + data: { + vaultId, + userId, + type: 'DEPOSIT', + status: 'PENDING', + amount: input.amount, + description: input.description, + reference, + idempotencyKey: input.idempotencyKey, + onChainVaultId: vault.onChainVaultId, + }, + }); + + await prisma.outboxEvent.create({ + data: { + data: { + eventType: OutboxEventType.VAULT_DEPOSIT, + aggregateId: transaction.id, + aggregateType: 'VaultTransaction', + payload: JSON.stringify({ + amount: input.amount, + reference, + type: 'DEPOSIT', + }), + }, + }); + + return transaction; }); const queue = getVaultQueue(); - await queue.add('vault-reconcile', { vaultTransactionId: transaction.id, type: 'DEPOSIT' }, { + await queue.add('vault-reconcile', { vaultTransactionId: result.id, type: 'DEPOSIT' }, { attempts: 5, backoff: { type: 'exponential', delay: 5000 }, removeOnComplete: true, removeOnFail: false, }); - return transaction; + return result; } /** @@ -184,39 +201,48 @@ export class VaultService { } const reference = generateReference('WTH'); - const transaction = await vaultTransactionRepository.create({ - vaultId, - userId, - type: 'WITHDRAWAL', - status: 'PENDING', - amount: input.amount, - description: input.description, - reference, - idempotencyKey: input.idempotencyKey, - onChainVaultId: vault.onChainVaultId, - }); - await vaultEventRepository.create({ - vaultId, - userId, - eventType: 'WITHDRAWAL', - status: 'REQUESTED', - payload: JSON.stringify({ - amount: input.amount, - reference, - type: 'WITHDRAWAL', - }), + const result = await prisma.$transaction(async (tx) => { + const transaction = await tx.vaultTransaction.create({ + data: { + vaultId, + userId, + type: 'WITHDRAWAL', + status: 'PENDING', + amount: input.amount, + description: input.description, + reference, + idempotencyKey: input.idempotencyKey, + onChainVaultId: vault.onChainVaultId, + }, + }); + + await prisma.outboxEvent.create({ + data: { + data: { + eventType: OutboxEventType.VAULT_WITHDRAWAL, + aggregateId: transaction.id, + aggregateType: 'VaultTransaction', + payload: JSON.stringify({ + amount: input.amount, + reference, + type: 'WITHDRAWAL', + }), + }, + }); + + return transaction; }); const queue = getVaultQueue(); - await queue.add('vault-reconcile', { vaultTransactionId: transaction.id, type: 'WITHDRAWAL' }, { + await queue.add('vault-reconcile', { vaultTransactionId: result.id, type: 'WITHDRAWAL' }, { attempts: 5, backoff: { type: 'exponential', delay: 5000 }, removeOnComplete: true, removeOnFail: false, }); - return transaction; + return result; } /** @@ -238,16 +264,31 @@ export class VaultService { unlocksAt, }); - await vaultEventRepository.create({ - vaultId, - userId, - eventType: 'LOCK', - status: 'CONFIRMED', - payload: JSON.stringify({ - lockPeriod: input.lockPeriod, - unlocksAt: unlocksAt.toISOString(), - }), - confirmedAt: new Date(), + await prisma.$transaction(async (tx) => { + await tx.vaultEvent.create({ + data: { + vaultId, + userId, + eventType: 'LOCK' as VaultEventType, + status: 'CONFIRMED' as StellarSubmissionStatus, + payload: JSON.stringify({ + lockPeriod: input.lockPeriod, + unlocksAt: unlocksAt.toISOString(), + }), + confirmedAt: new Date(), + }, + }); + + await prisma.outboxEvent.create({ + data: { + eventType: OutboxEventType.VAULT_LOCK, + aggregateId: vaultId, + aggregateType: 'SavingsVault', + payload: JSON.stringify({ + lockPeriod: input.lockPeriod, + unlocksAt: unlocksAt.toISOString(), + }), + }); }); return updated; @@ -271,13 +312,25 @@ export class VaultService { status: VaultStatus.ACTIVE, }); - await vaultEventRepository.create({ - vaultId, - userId, - eventType: 'UNLOCK', - status: 'CONFIRMED', - payload: JSON.stringify({ unlockedAt: new Date().toISOString() }), - confirmedAt: new Date(), + await prisma.$transaction(async (tx) => { + await tx.vaultEvent.create({ + data: { + vaultId, + userId, + eventType: 'UNLOCK' as VaultEventType, + status: 'CONFIRMED' as StellarSubmissionStatus, + payload: JSON.stringify({ unlockedAt: new Date().toISOString() }), + confirmedAt: new Date(), + }, + }); + + await prisma.outboxEvent.create({ + data: { + eventType: OutboxEventType.VAULT_UNLOCK, + aggregateId: vaultId, + aggregateType: 'SavingsVault', + payload: JSON.stringify({ unlockedAt: new Date().toISOString() }), + }); }); return updated; @@ -301,13 +354,25 @@ export class VaultService { status: VaultStatus.CLOSED, }); - await vaultEventRepository.create({ - vaultId, - userId, - eventType: 'CLOSE', - status: 'CONFIRMED', - payload: JSON.stringify({ closedAt: new Date().toISOString() }), - confirmedAt: new Date(), + await prisma.$transaction(async (tx) => { + await tx.vaultEvent.create({ + data: { + vaultId, + userId, + eventType: 'CLOSE' as VaultEventType, + status: 'CONFIRMED' as StellarSubmissionStatus, + payload: JSON.stringify({ closedAt: new Date().toISOString() }), + confirmedAt: new Date(), + }, + }); + + await prisma.outboxEvent.create({ + data: { + eventType: OutboxEventType.VAULT_CLOSE, + aggregateId: vaultId, + aggregateType: 'SavingsVault', + payload: JSON.stringify({ closedAt: new Date().toISOString() }), + }); }); return updated; diff --git a/Backend/tests/integration/outbox.integration.test.ts b/Backend/tests/integration/outbox.integration.test.ts new file mode 100644 index 0000000..d4950a7 --- /dev/null +++ b/Backend/tests/integration/outbox.integration.test.ts @@ -0,0 +1,288 @@ +import { prisma } from '../../src/database'; +import { outboxRepository } from '../../src/repositories/outbox.repository'; +import { OutboxEventStatus, OutboxEventType } from '@prisma/client'; + +jest.mock('../../src/queues', () => ({ + getOutboxProcessorQueue: jest.fn(() => ({ + add: jest.fn().mockResolvedValue({ id: 'job-1' }), + })), +})); + +jest.mock('../../src/jobs/outbox.processor', () => ({ + runOutboxProcessor: jest.fn(), + stopOutboxProcessor: jest.fn(), +})); + +describe('Outbox Integration', () => { + beforeAll(async () => { + await prisma.$connect(); + }); + + afterAll(async () => { + await prisma.$disconnect(); + }); + + beforeEach(async () => { + await prisma.outboxEvent.deleteMany({}); + }); + + afterEach(async () => { + await prisma.outboxEvent.deleteMany({}); + }); + + describe('Transactional outbox with database writes', () => { + it('persists outbox event when a database transaction succeeds', async () => { + const userId = 'test-user-' + Date.now(); + + await prisma.$transaction(async (tx) => { + await tx.user.create({ + data: { + id: userId, + email: `${userId}@example.com`, + passwordHash: 'hashed-password', + }, + }); + + await tx.outboxEvent.create({ + data: { + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: userId, + aggregateType: 'User', + payload: JSON.stringify({ + to: `${userId}@example.com`, + userId, + token: 'verification-token', + expiresAt: new Date(Date.now() + 86400000).toISOString(), + }), + }, + }); + }); + + const events = await outboxRepository.findPending(10); + expect(events.length).toBeGreaterThanOrEqual(1); + + const verificationEvent = events.find( + (e) => e.eventType === OutboxEventType.EMAIL_VERIFICATION + ); + expect(verificationEvent).toBeDefined(); + expect(verificationEvent?.aggregateId).toBe(userId); + expect(verificationEvent?.status).toBe(OutboxEventStatus.PENDING); + }); + + it('does not persist outbox event when a database transaction fails', async () => { + const userId = 'test-user-fail-' + Date.now(); + + await expect( + prisma.$transaction(async (tx) => { + await tx.user.create({ + data: { + id: userId, + email: `${userId}@example.com`, + passwordHash: 'hashed-password', + }, + }); + + await tx.outboxEvent.create({ + data: { + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: userId, + aggregateType: 'User', + payload: JSON.stringify({ + to: `${userId}@example.com`, + userId, + token: 'verification-token', + expiresAt: new Date(Date.now() + 86400000).toISOString(), + }), + }, + }); + + throw new Error('Simulated transaction failure'); + }) + ).rejects.toThrow('Simulated transaction failure'); + + const events = await prisma.outboxEvent.findMany({ + where: { aggregateId: userId }, + }); + expect(events.length).toBe(0); + }); + + it('preserves outbox events after a failed queue publication', async () => { + const userId = 'test-user-queue-fail-' + Date.now(); + + await prisma.$transaction(async (tx) => { + await tx.user.create({ + data: { + id: userId, + email: `${userId}@example.com`, + passwordHash: 'hashed-password', + }, + }); + + await tx.outboxEvent.create({ + data: { + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: userId, + aggregateType: 'User', + payload: JSON.stringify({ + to: `${userId}@example.com`, + userId, + token: 'verification-token', + expiresAt: new Date(Date.now() + 86400000).toISOString(), + }), + }, + }); + }); + + const eventsBefore = await prisma.outboxEvent.findMany({ + where: { aggregateId: userId }, + }); + expect(eventsBefore.length).toBe(1); + expect(eventsBefore[0].status).toBe(OutboxEventStatus.PENDING); + + await outboxRepository.markFailed(eventsBefore[0].id, 'Queue connection timeout'); + + const eventsAfter = await prisma.outboxEvent.findMany({ + where: { aggregateId: userId }, + }); + expect(eventsAfter.length).toBe(1); + expect(eventsAfter[0].status).toBe(OutboxEventStatus.FAILED); + expect(eventsAfter[0].attemptCount).toBe(1); + }); + + it('transitions event to DEAD_LETTER after exceeding max retry attempts', async () => { + const userId = 'test-user-dl-' + Date.now(); + + await prisma.$transaction(async (tx) => { + await tx.user.create({ + data: { + id: userId, + email: `${userId}@example.com`, + passwordHash: 'hashed-password', + }, + }); + + await tx.outboxEvent.create({ + data: { + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: userId, + aggregateType: 'User', + payload: JSON.stringify({ + to: `${userId}@example.com`, + userId, + token: 'verification-token', + expiresAt: new Date(Date.now() + 86400000).toISOString(), + }), + maxAttempts: 3, + }, + }); + }); + + const events = await prisma.outboxEvent.findMany({ + where: { aggregateId: userId }, + }); + const eventId = events[0].id; + + for (let i = 0; i < 3; i++) { + await outboxRepository.markFailed(eventId, `Attempt ${i + 1} failed`); + } + + const failedEvent = await prisma.outboxEvent.findUnique({ + where: { id: eventId }, + }); + expect(failedEvent?.status).toBe(OutboxEventStatus.FAILED); + expect(failedEvent?.attemptCount).toBe(3); + + await outboxRepository.markDeadLetter( + eventId, + 'Max retries exceeded: persistent queue failure' + ); + + const deadLetterEvent = await prisma.outboxEvent.findUnique({ + where: { id: eventId }, + }); + expect(deadLetterEvent?.status).toBe(OutboxEventStatus.DEAD_LETTER); + expect(deadLetterEvent?.deadLetterReason).toBe( + 'Max retries exceeded: persistent queue failure' + ); + }); + + it('successfully publishes an event and marks it as PUBLISHED', async () => { + const userId = 'test-user-publish-' + Date.now(); + + await prisma.$transaction(async (tx) => { + await tx.user.create({ + data: { + id: userId, + email: `${userId}@example.com`, + passwordHash: 'hashed-password', + }, + }); + + await tx.outboxEvent.create({ + data: { + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: userId, + aggregateType: 'User', + payload: JSON.stringify({ + to: `${userId}@example.com`, + userId, + token: 'verification-token', + expiresAt: new Date(Date.now() + 86400000).toISOString(), + }), + }, + }); + }); + + const events = await prisma.outboxEvent.findMany({ + where: { aggregateId: userId }, + }); + const eventId = events[0].id; + + await outboxRepository.markPublished(eventId); + + const publishedEvent = await prisma.outboxEvent.findUnique({ + where: { id: eventId }, + }); + expect(publishedEvent?.status).toBe(OutboxEventStatus.PUBLISHED); + expect(publishedEvent?.publishedAt).toBeDefined(); + expect(publishedEvent?.nextRetryAt).toBeNull(); + }); + }); + + describe('Idempotency', () => { + it('prevents duplicate outbox events for the same aggregate and event type', async () => { + const userId = 'test-user-idem-' + Date.now(); + + await prisma.$transaction(async (tx) => { + await tx.user.create({ + data: { + id: userId, + email: `${userId}@example.com`, + passwordHash: 'hashed-password', + }, + }); + + await tx.outboxEvent.create({ + data: { + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: userId, + aggregateType: 'User', + payload: JSON.stringify({ + to: `${userId}@example.com`, + userId, + token: 'verification-token', + expiresAt: new Date(Date.now() + 86400000).toISOString(), + }), + }, + }); + }); + + const existingEvent = await outboxRepository.findByIdempotencyKey( + userId, + OutboxEventType.EMAIL_VERIFICATION + ); + expect(existingEvent).toBeDefined(); + expect(existingEvent?.aggregateId).toBe(userId); + }); + }); +}); \ No newline at end of file diff --git a/Backend/tests/unit/outbox.repository.test.ts b/Backend/tests/unit/outbox.repository.test.ts new file mode 100644 index 0000000..5580187 --- /dev/null +++ b/Backend/tests/unit/outbox.repository.test.ts @@ -0,0 +1,243 @@ +import { outboxRepository } from '../../src/repositories/outbox.repository'; +import { OutboxEventStatus, OutboxEventType } from '@prisma/client'; + +jest.mock('../../src/database', () => ({ + prisma: { + outboxEvent: { + create: jest.fn(), + findUnique: jest.fn(), + findMany: jest.fn(), + findFirst: jest.fn(), + update: jest.fn(), + updateMany: jest.fn(), + deleteMany: jest.fn(), + }, + }, +})); + +const mockPrisma = require('../../src/database').prisma; + +describe('OutboxRepository', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + describe('create', () => { + it('creates an outbox event with the given data', async () => { + const mockEvent = { + id: 'event-1', + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: 'user-1', + aggregateType: 'User', + payload: JSON.stringify({ to: 'test@example.com' }), + attemptCount: 0, + status: OutboxEventStatus.PENDING, + createdAt: new Date(), + }; + mockPrisma.outboxEvent.create.mockResolvedValue(mockEvent); + + const result = await outboxRepository.create({ + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: 'user-1', + aggregateType: 'User', + payload: JSON.stringify({ to: 'test@example.com' }), + }); + + expect(mockPrisma.outboxEvent.create).toHaveBeenCalledWith({ + data: { + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: 'user-1', + aggregateType: 'User', + payload: JSON.stringify({ to: 'test@example.com' }), + }, + }); + expect(result).toEqual(mockEvent); + }); + }); + + describe('findPending', () => { + it('returns pending events that are due for retry', async () => { + const mockEvents = [ + { + id: 'event-1', + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: 'user-1', + aggregateType: 'User', + payload: '{}', + attemptCount: 0, + status: OutboxEventStatus.PENDING, + nextRetryAt: null, + }, + ]; + mockPrisma.outboxEvent.findMany.mockResolvedValue(mockEvents); + + const result = await outboxRepository.findPending(10); + + expect(mockPrisma.outboxEvent.findMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ + status: OutboxEventStatus.PENDING, + attemptCount: { lt: 5 }, + }), + orderBy: { createdAt: 'asc' }, + take: 10, + }) + ); + expect(result).toEqual(mockEvents); + }); + + it('returns empty array when no pending events exist', async () => { + mockPrisma.outboxEvent.findMany.mockResolvedValue([]); + + const result = await outboxRepository.findPending(10); + + expect(result).toEqual([]); + }); + }); + + describe('findByIdempotencyKey', () => { + it('returns the most recent pending or published event for the same aggregate and type', async () => { + const mockEvent = { + id: 'event-1', + eventType: OutboxEventType.EMAIL_VERIFICATION, + aggregateId: 'user-1', + aggregateType: 'User', + payload: '{}', + status: OutboxEventStatus.PENDING, + }; + mockPrisma.outboxEvent.findFirst.mockResolvedValue(mockEvent); + + const result = await outboxRepository.findByIdempotencyKey('user-1', OutboxEventType.EMAIL_VERIFICATION); + + expect(mockPrisma.outboxEvent.findFirst).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ + aggregateId: 'user-1', + eventType: OutboxEventType.EMAIL_VERIFICATION, + status: { + in: [OutboxEventStatus.PENDING, OutboxEventStatus.PUBLISHED], + }, + }), + orderBy: { createdAt: 'desc' }, + }) + ); + expect(result).toEqual(mockEvent); + }); + }); + + describe('markPublished', () => { + it('updates the event status to PUBLISHED with a timestamp', async () => { + const mockEvent = { + id: 'event-1', + status: OutboxEventStatus.PUBLISHED, + publishedAt: new Date(), + }; + mockPrisma.outboxEvent.update.mockResolvedValue(mockEvent); + + const result = await outboxRepository.markPublished('event-1'); + + expect(mockPrisma.outboxEvent.update).toHaveBeenCalledWith({ + where: { id: 'event-1' }, + data: expect.objectContaining({ + status: OutboxEventStatus.PUBLISHED, + publishedAt: expect.any(Date), + nextRetryAt: null, + }), + }); + expect(result).toEqual(mockEvent); + }); + }); + + describe('markFailed', () => { + it('increments attempt count and sets FAILED status with retry timestamp', async () => { + const mockEvent = { + id: 'event-1', + status: OutboxEventStatus.FAILED, + attemptCount: 1, + nextRetryAt: new Date(Date.now() + 60000), + }; + mockPrisma.outboxEvent.update.mockResolvedValue(mockEvent); + + const result = await outboxRepository.markFailed('event-1', 'Publish failed'); + + expect(mockPrisma.outboxEvent.update).toHaveBeenCalledWith({ + where: { id: 'event-1' }, + data: expect.objectContaining({ + attemptCount: { increment: 1 }, + status: OutboxEventStatus.FAILED, + nextRetryAt: expect.any(Date), + failedAt: expect.any(Date), + deadLetterReason: 'Publish failed', + }), + }); + expect(result).toEqual(mockEvent); + }); + }); + + describe('markDeadLetter', () => { + it('sets the event status to DEAD_LETTER with a reason', async () => { + const mockEvent = { + id: 'event-1', + status: OutboxEventStatus.DEAD_LETTER, + deadLetterReason: 'Max retries exceeded', + }; + mockPrisma.outboxEvent.update.mockResolvedValue(mockEvent); + + const result = await outboxRepository.markDeadLetter('event-1', 'Max retries exceeded'); + + expect(mockPrisma.outboxEvent.update).toHaveBeenCalledWith({ + where: { id: 'event-1' }, + data: expect.objectContaining({ + status: OutboxEventStatus.DEAD_LETTER, + deadLetterReason: 'Max retries exceeded', + failedAt: expect.any(Date), + }), + }); + expect(result).toEqual(mockEvent); + }); + }); + + describe('resetFailedEvents', () => { + it('resets failed events that are within retry budget and past their retry time', async () => { + const mockResult = { count: 3 }; + mockPrisma.outboxEvent.updateMany.mockResolvedValue(mockResult); + + const result = await outboxRepository.resetFailedEvents(); + + expect(mockPrisma.outboxEvent.updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ + status: OutboxEventStatus.FAILED, + attemptCount: { lt: 5 }, + nextRetryAt: { lte: expect.any(Date) }, + }), + data: expect.objectContaining({ + status: OutboxEventStatus.PENDING, + nextRetryAt: null, + }), + }) + ); + expect(result).toEqual(mockResult); + }); + }); + + describe('deletePublishedOlderThan', () => { + it('deletes published events older than the cutoff date', async () => { + const mockResult = { count: 50 }; + mockPrisma.outboxEvent.deleteMany.mockResolvedValue(mockResult); + + const cutoff = new Date('2024-01-01'); + const result = await outboxRepository.deletePublishedOlderThan(cutoff); + + expect(mockPrisma.outboxEvent.deleteMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ + status: OutboxEventStatus.PUBLISHED, + publishedAt: { lt: cutoff }, + }), + }) + ); + expect(result).toEqual(mockResult); + }); + }); +}); \ No newline at end of file