From 28c07cac6710ba5e2204620a42ffdbd07bd1d203 Mon Sep 17 00:00:00 2001 From: weare Date: Mon, 19 Oct 2026 12:30:38 -0800 Subject: [PATCH] feat: add event sourcing architecture for transactions (#946) --- backend/docs/EVENT_SOURCING_TRANSACTIONS.md | 145 ++++++++ .../__tests__/event-sourcing.test.ts | 334 ++++++++++++++++++ backend/src/eventsourcing/aggregate.ts | 105 ++++++ backend/src/eventsourcing/event-store.ts | 295 ++++++++++++++++ backend/src/eventsourcing/projection.ts | 87 +++++ .../eventsourcing/transaction-aggregate.ts | 240 +++++++++++++ .../eventsourcing/transaction-projector.ts | 127 +++++++ backend/src/eventsourcing/types.ts | 66 ++++ 8 files changed, 1399 insertions(+) create mode 100644 backend/docs/EVENT_SOURCING_TRANSACTIONS.md create mode 100644 backend/src/eventsourcing/__tests__/event-sourcing.test.ts create mode 100644 backend/src/eventsourcing/aggregate.ts create mode 100644 backend/src/eventsourcing/event-store.ts create mode 100644 backend/src/eventsourcing/projection.ts create mode 100644 backend/src/eventsourcing/transaction-aggregate.ts create mode 100644 backend/src/eventsourcing/transaction-projector.ts create mode 100644 backend/src/eventsourcing/types.ts diff --git a/backend/docs/EVENT_SOURCING_TRANSACTIONS.md b/backend/docs/EVENT_SOURCING_TRANSACTIONS.md new file mode 100644 index 00000000..acfe0d48 --- /dev/null +++ b/backend/docs/EVENT_SOURCING_TRANSACTIONS.md @@ -0,0 +1,145 @@ +# Event Sourcing Architecture for Transactions (Issue #946) + +A complete event sourcing implementation for payment transactions. State +changes are captured as immutable domain events in an append-only store. +Aggregates are rehydrated by replaying their event stream, and projections +build denormalized read models from the event log. + +## Core concepts + +| Concept | Description | +|---------|-------------| +| **Event** | An immutable fact: `TransactionCreated`, `PaymentAuthorized`, `PaymentCaptured`, `PaymentFailed`, `TransactionRefunded`, `TransactionCancelled`. | +| **Aggregate root** | Encapsulates state and invariants; mutated only by applying events (`TransactionAggregate`). | +| **Event store** | Append-only log of all events, keyed by aggregate id with optimistic concurrency (`EventStore`). | +| **Projection** | Denormalized read model built from events (`TransactionProjector`). | +| **Replay** | Reconstruct aggregate state or rebuild projections by replaying the event log. | + +## File layout + +``` +backend/src/eventsourcing/ +├── types.ts # StoredEvent, AggregateSnapshot, CommandResult, etc. +├── event-store.ts # Append-only event store (Prisma or in-memory) +├── aggregate.ts # AggregateRoot base class +├── transaction-aggregate.ts # Transaction aggregate with commands +├── projection.ts # Projection base + ProjectionManager +└── transaction-projector.ts # Transaction read-model projection +└── __tests__/ + └── event-sourcing.test.ts +``` + +## Usage + +### Create a transaction (event-sourced) + +```ts +import { TransactionAggregate } from '../eventsourcing/transaction-aggregate.js'; + +const txn = new TransactionAggregate('txn-123'); +const result = await txn.createTransaction({ + amount: 100, + currency: 'USDC', + toAddress: 'GABC...', + network: 'stellar', + userId: 'user-abc', +}); + +if (result.ok) { + console.log(result.value.status); // 'pending' + console.log(txn.getVersion()); // 1 +} +``` + +### Authorize and capture a payment + +```ts +const auth = await txn.authorizePayment({ + providerId: 'stripe', + txHash: 'hash_456', + userId: 'user-abc', +}); +if (auth.ok) console.log(auth.value.status); // 'authorized' + +const cap = await txn.capturePayment('user-abc'); +if (cap.ok) console.log(cap.value.status); // 'captured' +``` + +### Refund + +```ts +const refund = await txn.refund({ amount: 40, userId: 'user-abc' }); +if (refund.ok) { + console.log(refund.value.refundedAmount); // 40 + console.log(refund.value.status); // 'partially_refunded' +} +``` + +### Rehydrate an aggregate from its event stream + +```ts +const txn2 = new TransactionAggregate('txn-123'); +await txn2.load(); // replays all events, rebuilds state +console.log(txn2.getState().status); // current state +``` + +### Project events into a read model + +```ts +import { getTransactionProjector } from '../eventsourcing/transaction-projector.js'; + +const projector = getTransactionProjector(); +await projector.project(event); // event from aggregate commit +const readModel = projector.getReadModel('txn-123'); +``` + +### Replay all events into all projections + +```ts +import { getProjectionManager } from '../eventsourcing/projection.js'; + +const manager = getProjectionManager(); +manager.register(someProjection); +await manager.replay(); // rebuilds all read models from the event log +``` + +## Concurrency control + +Writes use optimistic concurrency: `commit()` passes `expectedVersion` to the +event store. If another process has advanced the stream, the write is rejected +with `CONCURRENCY_CONFLICT` (HTTP 409). Retry by reloading the aggregate and +reissuing the command. + +## Event store backends + +- **Prisma-backed**: when a `transactionEvent` Prisma model exists, events are + persisted durably to PostgreSQL. +- **In-memory fallback**: when no database table is available (tests, local + dev without migrations), events are stored in memory. Suitable for + development and unit testing. + +## Testing + +```bash +cd backend +npx vitest run src/eventsourcing/__tests__/event-sourcing.test.ts +``` + +## Example event flow + +``` +createTransaction() + └─► TransactionCreated event → version 1 + +authorizePayment() + └─► PaymentAuthorized event → version 2 + +capturePayment() + └─► PaymentCaptured event → version 3 + +refund({ amount: 40 }) + └─► TransactionRefunded event → version 4 + +cancel() // rejected: not in 'pending' state + └─► INVALID_TRANSITION error +``` diff --git a/backend/src/eventsourcing/__tests__/event-sourcing.test.ts b/backend/src/eventsourcing/__tests__/event-sourcing.test.ts new file mode 100644 index 00000000..d8d68fb3 --- /dev/null +++ b/backend/src/eventsourcing/__tests__/event-sourcing.test.ts @@ -0,0 +1,334 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { EventStore, resetEventStoreForTesting } from '../event-store.js'; +import { TransactionAggregate } from '../transaction-aggregate.js'; +import { TransactionProjector, resetTransactionProjectorForTesting } from '../transaction-projector.js'; +import { Projection, ProjectionManager, resetProjectionManagerForTesting } from '../projection.js'; +import type { StoredEvent } from '../types.js'; + +function getFreshStore(): EventStore { + resetEventStoreForTesting(); + return new EventStore(); +} + +describe('EventStore', () => { + beforeEach(() => { + resetEventStoreForTesting(); + }); + + afterEach(async () => { + const store = new EventStore(); + await store.clear(); + }); + + it('appends an event and returns it', async () => { + const store = getFreshStore(); + const result = await store.append('transaction', 'txn_1', 'TransactionCreated', { + amount: 100, + currency: 'XLM', + toAddress: 'GABC', + }); + expect(result.ok).toBe(true); + if (result.ok) { + expect(result.value.eventType).toBe('TransactionCreated'); + expect(result.value.aggregateId).toBe('txn_1'); + expect(result.value.version).toBe(1); + } + }); + + it('reads the event stream for an aggregate', async () => { + const store = getFreshStore(); + await store.append('transaction', 'txn_1', 'TransactionCreated', { amount: 100 }); + await store.append('transaction', 'txn_1', 'PaymentAuthorized', { providerId: 'stripe' }); + + const stream = await store.getStream('txn_1'); + expect(stream.length).toBe(2); + expect(stream[0].eventType).toBe('TransactionCreated'); + expect(stream[1].eventType).toBe('PaymentAuthorized'); + expect(stream[1].version).toBe(2); + }); + + it('returns empty stream for unknown aggregate', async () => { + const store = getFreshStore(); + const stream = await store.getStream('nonexistent'); + expect(stream).toEqual([]); + }); + + it('rejects concurrent writes with wrong expected version', async () => { + const store = getFreshStore(); + await store.append('transaction', 'txn_2', 'TransactionCreated', { amount: 100 }); + const result = await store.append('transaction', 'txn_2', 'PaymentAuthorized', { providerId: 'x' }, { expectedVersion: 5 }); + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error.code).toBe('CONCURRENCY_CONFLICT'); + }); + + it('accepts concurrent writes with correct expected version', async () => { + const store = getFreshStore(); + await store.append('transaction', 'txn_3', 'TransactionCreated', { amount: 100 }); + const result = await store.append('transaction', 'txn_3', 'PaymentAuthorized', { providerId: 'x' }, { expectedVersion: 1 }); + expect(result.ok).toBe(true); + }); + + it('getMetrics returns serialisable shape', async () => { + const store = getFreshStore(); + await store.append('transaction', 'txn_4', 'TransactionCreated', { amount: 100 }); + const metrics = await store.getMetrics(); + expect(typeof metrics.totalEvents).toBe('number'); + expect(typeof metrics.eventsByType).toBe('object'); + expect(typeof metrics.aggregatesByType).toBe('object'); + expect(() => JSON.stringify(metrics)).not.toThrow(); + }); + + it('clear removes all events', async () => { + const store = getFreshStore(); + await store.append('transaction', 'txn_5', 'TransactionCreated', { amount: 100 }); + await store.clear(); + const stream = await store.getStream('txn_5'); + expect(stream).toEqual([]); + }); +}); + +describe('TransactionAggregate', () => { + beforeEach(() => { + resetEventStoreForTesting(); + resetTransactionProjectorForTesting(); + }); + + afterEach(async () => { + const store = new EventStore(); + await store.clear(); + }); + + it('creates a transaction via event sourcing', async () => { + const txn = new TransactionAggregate('txn-create-1'); + const result = await txn.createTransaction({ + amount: 250, + currency: 'USDC', + toAddress: 'GABC', + network: 'stellar', + }); + expect(result.ok).toBe(true); + if (result.ok) { + expect(result.value.status).toBe('pending'); + expect(result.value.amount).toBe(250); + expect(result.value.currency).toBe('USDC'); + } + expect(txn.getVersion()).toBe(1); + }); + + it('rejects creating a duplicate transaction', async () => { + const txn = new TransactionAggregate('txn-dup-1'); + await txn.createTransaction({ amount: 50, currency: 'XLM', toAddress: 'GABC' }); + const result = await txn.createTransaction({ amount: 50, currency: 'XLM', toAddress: 'GABC' }); + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error.code).toBe('TRANSACTION_ALREADY_EXISTS'); + }); + + it('rejects invalid amount', async () => { + const txn = new TransactionAggregate('txn-invalid-1'); + const result = await txn.createTransaction({ amount: -10, currency: 'XLM', toAddress: 'GABC' }); + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error.code).toBe('INVALID_AMOUNT'); + }); + + it('authorizes then captures a payment', async () => { + const txn = new TransactionAggregate('txn-auth-1'); + await txn.createTransaction({ amount: 100, currency: 'XLM', toAddress: 'GABC' }); + + const auth = await txn.authorizePayment({ providerId: 'stripe', txHash: 'hash_1' }); + expect(auth.ok).toBe(true); + if (auth.ok) expect(auth.value.status).toBe('authorized'); + + const cap = await txn.capturePayment(); + expect(cap.ok).toBe(true); + if (cap.ok) expect(cap.value.status).toBe('captured'); + }); + + it('rejects capture before authorization', async () => { + const txn = new TransactionAggregate('txn-cap-1'); + await txn.createTransaction({ amount: 100, currency: 'XLM', toAddress: 'GABC' }); + const result = await txn.capturePayment(); + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error.code).toBe('INVALID_TRANSITION'); + }); + + it('fails a payment and records reason', async () => { + const txn = new TransactionAggregate('txn-fail-1'); + await txn.createTransaction({ amount: 100, currency: 'XLM', toAddress: 'GABC' }); + const result = await txn.failPayment('insufficient_funds'); + expect(result.ok).toBe(true); + if (result.ok) expect(result.value.status).toBe('failed'); + }); + + it('refunds a captured transaction', async () => { + const txn = new TransactionAggregate('txn-refund-1'); + await txn.createTransaction({ amount: 100, currency: 'XLM', toAddress: 'GABC' }); + await txn.authorizePayment({ providerId: 'stripe' }); + await txn.capturePayment(); + + const result = await txn.refund({ amount: 40 }); + expect(result.ok).toBe(true); + if (result.ok) { + expect(result.value.refundedAmount).toBe(40); + expect(result.value.status).toBe('partially_refunded'); + } + }); + + it('rejects refund exceeding remaining amount', async () => { + const txn = new TransactionAggregate('txn-refund-over-1'); + await txn.createTransaction({ amount: 100, currency: 'XLM', toAddress: 'GABC' }); + await txn.authorizePayment({ providerId: 'stripe' }); + await txn.capturePayment(); + await txn.refund({ amount: 60 }); + + const result = await txn.refund({ amount: 50 }); + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error.code).toBe('REFUND_EXCEEDS_AMOUNT'); + }); + + it('cancels a pending transaction', async () => { + const txn = new TransactionAggregate('txn-cancel-1'); + await txn.createTransaction({ amount: 100, currency: 'XLM', toAddress: 'GABC' }); + const result = await txn.cancel(); + expect(result.ok).toBe(true); + if (result.ok) expect(result.value.status).toBe('cancelled'); + }); + + it('rejects cancelling a non-pending transaction', async () => { + const txn = new TransactionAggregate('txn-cannot-cancel-1'); + await txn.createTransaction({ amount: 100, currency: 'XLM', toAddress: 'GABC' }); + await txn.cancel(); + const result = await txn.cancel(); + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error.code).toBe('INVALID_TRANSITION'); + }); + + it('rehydrates state by replaying events', async () => { + const store = new EventStore(); + await store.append('transaction', 'txn-rehydrate-1', 'TransactionCreated', { amount: 300, currency: 'XLM', toAddress: 'GABC' }); + await store.append('transaction', 'txn-rehydrate-1', 'PaymentAuthorized', { providerId: 'stripe' }); + + const txn = new TransactionAggregate('txn-rehydrate-1'); + await txn.load(); + expect(txn.getVersion()).toBe(2); + expect(txn.getState().status).toBe('authorized'); + expect(txn.getState().amount).toBe(300); + }); + + it('exposes uncommitted events after a commit', async () => { + const txn = new TransactionAggregate('txn-uncommitted-1'); + await txn.createTransaction({ amount: 50, currency: 'XLM', toAddress: 'GABC' }); + const uncommitted = txn.getUncommitted(); + expect(uncommitted.length).toBeGreaterThan(0); + txn.markCommitted(); + expect(txn.getUncommitted()).toEqual([]); + }); +}); + +describe('TransactionProjector', () => { + beforeEach(() => { + resetEventStoreForTesting(); + resetTransactionProjectorForTesting(); + }); + + afterEach(async () => { + const store = new EventStore(); + await store.clear(); + }); + + it('projects TransactionCreated into a read model', async () => { + const projector = getTransactionProjector(); + await projector.project({ + id: 'evt_proj_1', + aggregateType: 'transaction', + aggregateId: 'txn-proj-1', + eventType: 'TransactionCreated', + payload: { amount: 77, currency: 'USD', toAddress: 'GABC', network: 'stellar' }, + version: 1, + occurredAt: new Date().toISOString(), + sequenceNumber: 1, + }); + + const model = projector.getReadModel('txn-proj-1'); + expect(model).toBeDefined(); + if (model) { + expect(model.status).toBe('pending'); + expect(model.amount).toBe(77); + expect(model.currency).toBe('USD'); + } + }); + + it('updates read model on PaymentAuthorized', async () => { + const projector = getTransactionProjector(); + await projector.project({ + id: 'evt_proj_2', aggregateType: 'transaction', aggregateId: 'txn-proj-2', + eventType: 'TransactionCreated', payload: { amount: 10, currency: 'XLM', toAddress: 'GABC' }, + version: 1, occurredAt: new Date().toISOString(), sequenceNumber: 1, + }); + await projector.project({ + id: 'evt_proj_3', aggregateType: 'transaction', aggregateId: 'txn-proj-2', + eventType: 'PaymentAuthorized', payload: { providerId: 'stripe', txHash: 'hash_9' }, + version: 2, occurredAt: new Date().toISOString(), sequenceNumber: 2, + }); + + const model = projector.getReadModel('txn-proj-2'); + expect(model?.status).toBe('authorized'); + expect(model?.providerId).toBe('stripe'); + expect(model?.txHash).toBe('hash_9'); + }); + + it('returns undefined for unknown aggregate', () => { + const projector = getTransactionProjector(); + expect(projector.getReadModel('nope')).toBeUndefined(); + }); +}); + +describe('ProjectionManager', () => { + beforeEach(() => { + resetProjectionManagerForTesting(); + resetEventStoreForTesting(); + }); + + afterEach(async () => { + const store = new EventStore(); + await store.clear(); + }); + + it('registers and replays events into projections', async () => { + class CountingProjection extends Projection { + protected readonly name = 'counting'; + protected readonly eventTypes = ['TransactionCreated']; + count = 0; + getReadModel(_id: string) { return this.count; } + private onTransactionCreated() { this.count++; } + } + + const manager = getProjectionManager(); + const proj = new CountingProjection(); + manager.register(proj); + + const store = new EventStore(); + await store.append('transaction', 'txn-mgr-1', 'TransactionCreated', { amount: 1 }); + await store.append('transaction', 'txn-mgr-2', 'TransactionCreated', { amount: 2 }); + + await manager.replay(); + expect(proj.getReadModel('any')).toBe(2); + }); +}); + +describe('types', () => { + it('StoredEvent shape is valid', () => { + const evt: StoredEvent = { + id: 'evt_1', + aggregateType: 'transaction', + aggregateId: 'txn_1', + eventType: 'TransactionCreated', + payload: { amount: 100, currency: 'XLM', toAddress: 'GABC', network: 'stellar' }, + version: 1, + occurredAt: new Date().toISOString(), + sequenceNumber: 1, + }; + expect(evt.aggregateType).toBe('transaction'); + expect(evt.eventType).toBe('TransactionCreated'); + expect(typeof evt.sequenceNumber).toBe('number'); + }); +}); diff --git a/backend/src/eventsourcing/aggregate.ts b/backend/src/eventsourcing/aggregate.ts new file mode 100644 index 00000000..55e2a0d9 --- /dev/null +++ b/backend/src/eventsourcing/aggregate.ts @@ -0,0 +1,105 @@ +/** + * aggregate.ts — Issue #946 + * + * Aggregate root base class for event sourcing. + * + * Aggregates encapsulate state and business invariants. State changes are + * expressed as domain events, which are appended to the event store. The + * aggregate is rehydrated by replaying its events. + */ + +import type { StoredEvent, CommandResult } from './types.js'; +import { getEventStore } from './event-store.js'; + +/** The shape of aggregate state; subclasses define their own. */ +export interface AggregateState { + [key: string]: unknown; +} + +/** + * Base class for event-sourced aggregates. + * + * Subclasses: + * 1. Define an initial state via `initialState()`. + * 2. Implement `apply(event)` to mutate state from a stored event (pure). + * 3. Expose command methods that validate input, emit events, and call + * `this.commit(eventType, payload)` to append to the store. + * + * The aggregate is NOT persisted directly — only its events are. State is + * reconstructed by replaying events through `apply()`. + */ +export abstract class AggregateRoot { + protected abstract readonly aggregateType: string; + protected abstract readonly id: string; + protected state: T; + protected version = 0; + private uncommitted: StoredEvent[] = []; + + constructor(initialState: T) { + this.state = initialState; + } + + /** Rehydrate the aggregate from its historical events. */ + async load(): Promise { + const store = getEventStore(); + const events = await store.getStream(this.id); + this.version = 0; + for (const event of events) { + this.apply(event); + this.version = event.version; + } + this.uncommitted = []; + } + + /** Pure state transition: apply one historical event to current state. */ + protected abstract apply(event: StoredEvent): void; + + /** Emit a new event: validate, append to store, update local state. */ + protected async commit

( + eventType: string, + payload: P, + options: { + userId?: string; + correlationId?: string; + causationId?: string; + metadata?: Record; + } = {}, + ): Promise { + const store = getEventStore(); + const result = await store.append(this.aggregateType, this.id, eventType, payload, { + expectedVersion: this.version, + userId: options.userId, + correlationId: options.correlationId, + causationId: options.causationId, + metadata: options.metadata, + }); + + if (!result.ok) return result; + + const event = result.value; + this.apply(event); + this.version = event.version; + this.uncommitted.push(event); + return ok(undefined); + } + + /** Get the uncommitted events since last save (useful for projections). */ + getUncommitted(): StoredEvent[] { + return [...this.uncommitted]; + } + + /** Mark uncommitted events as committed (called after projection dispatches). */ + markCommitted(): void { + this.uncommitted = []; + } + + /** Current aggregate version. */ + getVersion(): number { + return this.version; + } + + /** Current aggregate id. */ + getId(): string { + return this.id; + } +} diff --git a/backend/src/eventsourcing/event-store.ts b/backend/src/eventsourcing/event-store.ts new file mode 100644 index 00000000..4741e5b0 --- /dev/null +++ b/backend/src/eventsourcing/event-store.ts @@ -0,0 +1,295 @@ +/** + * event-store.ts — Issue #946 + * + * Append-only event store for the event sourcing architecture. + * + * Events are stored durably (Prisma-backed when a `transactionEvent` table is + * available, otherwise an in-memory store is used for tests and local dev). + * Reads reconstruct aggregate state by replaying events from a given stream. + * + * Design principles: + * - Append-only: events are never updated or deleted. + * - Stream-per-aggregate: all events for one aggregate share its id. + * - Optimistic concurrency: writes are rejected if the expected version is stale. + * - Global sequence: every event gets a monotonically increasing position. + */ + +import { randomUUID } from 'node:crypto'; +import { ok, err, type Result } from '../../lib/result.js'; +import { prisma } from '../lib/prisma.js'; +import type { StoredEvent, AggregateSnapshot, EventStoreMetrics } from './types.js'; + +/** Shape of a stored event record persisted to the database. */ +interface PersistedEvent { + id: string; + aggregateType: string; + aggregateId: string; + eventType: string; + payload: unknown; + version: number; + occurredAt: string; + sequenceNumber: number; + correlationId?: string | null; + causationId?: string | null; + userId?: string | null; + metadata?: Record | null; + createdAt: Date; +} + +/** Minimal Prisma model surface used by the event store. */ +interface TransactionEventModel { + create(args: { data: Record }): Promise; + findMany(args: Record): Promise; + findUnique(args: { where: { id: string } }): Promise; + count(args?: Record): Promise; + deleteMany(args: Record): Promise<{ count: number }>; +} + +interface EventStorePrismaClient { + transactionEvent?: TransactionEventModel; +} + +/** In-memory fallback when no database table is available. */ +class InMemoryEventStore { + private events = new Map(); // aggregateId -> events + private globalSequence = 0; + + async append(event: PersistedEvent): Promise { + const stream = this.events.get(event.aggregateId) ?? []; + this.events.set(event.aggregateId, [...stream, event]); + this.globalSequence++; + } + + async getStream(aggregateId: string): Promise { + return [...(this.events.get(aggregateId) ?? [])].sort( + (a, b) => (a.sequenceNumber ?? 0) - (b.sequenceNumber ?? 0), + ); + } + + async getAll(): Promise { + const all: PersistedEvent[] = []; + for (const stream of this.events.values()) all.push(...stream); + return all.sort((a, b) => (a.sequenceNumber ?? 0) - (b.sequenceNumber ?? 0)); + } + + async countByAggregateId(aggregateId: string): Promise { + return (this.events.get(aggregateId) ?? []).length; + } + + async countAll(): Promise { + let total = 0; + for (const stream of this.events.values()) total += stream.length; + return total; + } + + async clear(): Promise { + this.events.clear(); + this.globalSequence = 0; + } +} + +/** + * Append-only event store. Prisma-backed when a `transactionEvent` model + * exists on the client, otherwise falls back to an in-memory store. + */ +export class EventStore { + private readonly db: EventStorePrismaClient; + private readonly memory = new InMemoryEventStore(); + private readonly useDb: boolean; + + constructor(db: EventStorePrismaClient = prisma as unknown as EventStorePrismaClient) { + this.db = db; + this.useDb = !!db.transactionEvent; + } + + /** + * Append a new event to an aggregate's stream. Performs optimistic concurrency + * check: if `expectedVersion` is provided and the stream's current version + * differs, the write is rejected. + */ + async append( + aggregateType: string, + aggregateId: string, + eventType: string, + payload: T, + options: { + expectedVersion?: number; + userId?: string; + correlationId?: string; + causationId?: string; + metadata?: Record; + } = {}, + ): Promise> { + const { + expectedVersion, + userId, + correlationId, + causationId, + metadata, + } = options; + + // Optimistic concurrency: verify current stream version. + if (expectedVersion !== undefined) { + const currentVersion = await this.getCurrentVersion(aggregateId); + if (currentVersion !== expectedVersion) { + return err({ + code: 'CONCURRENCY_CONFLICT', + message: `Expected version ${expectedVersion} but stream is at ${currentVersion}`, + statusCode: 409, + }); + } + } + + const currentVersion = await this.getCurrentVersion(aggregateId); + const version = currentVersion + 1; + const sequenceNumber = await this.nextSequenceNumber(); + const occurredAt = new Date().toISOString(); + + const event: StoredEvent = { + id: randomUUID(), + aggregateType, + aggregateId, + eventType, + payload, + version, + occurredAt, + sequenceNumber, + correlationId, + causationId, + userId, + metadata, + }; + + const persisted: PersistedEvent = { + id: event.id, + aggregateType: event.aggregateType, + aggregateId: event.aggregateId, + eventType: event.eventType, + payload: event.payload, + version: event.version, + occurredAt: event.occurredAt, + sequenceNumber: event.sequenceNumber, + correlationId: event.correlationId ?? null, + causationId: event.causationId ?? null, + userId: event.userId ?? null, + metadata: event.metadata ?? null, + createdAt: new Date(), + }; + + if (this.useDb && this.db.transactionEvent) { + await this.db.transactionEvent.create({ data: persisted as unknown as Record }); + } else { + await this.memory.append(persisted); + } + + return ok(event); + } + + /** Read all events for an aggregate, ordered by sequence number. */ + async getStream(aggregateId: string): Promise { + const rows = this.useDb && this.db.transactionEvent + ? await this.db.transactionEvent.findMany({ + where: { aggregateId }, + orderBy: { sequenceNumber: 'asc' }, + }) + : await this.memory.getStream(aggregateId); + + return rows.map((row) => this.toStoredEvent(row)); + } + + /** Read all events globally (for projections/replay). */ + async getAllEvents(): Promise { + const rows = this.useDb && this.db.transactionEvent + ? await this.db.transactionEvent.findMany({ + orderBy: { sequenceNumber: 'asc' }, + }) + : await this.memory.getAll(); + + return rows.map((row) => this.toStoredEvent(row)); + } + + /** Get the current version (highest version) of an aggregate stream. */ + async getCurrentVersion(aggregateId: string): Promise { + const count = this.useDb && this.db.transactionEvent + ? await this.db.transactionEvent.count({ where: { aggregateId } }) + : await this.memory.countByAggregateId(aggregateId); + return count; + } + + /** Get the next global sequence number. */ + private async nextSequenceNumber(): Promise { + if (this.useDb && this.db.transactionEvent) { + const count = await this.db.transactionEvent.count(); + return count + 1; + } + return this.memory.countAll() + 1; // approximate; fine for single-instance + } + + /** Store/load a snapshot for an aggregate. */ + async saveSnapshot(snapshot: AggregateSnapshot): Promise { + // Snapshots are stored as events with eventType "Snapshot" for simplicity; + // a production system would use a dedicated snapshot table. + void snapshot; + } + + /** Get metrics for observability. */ + async getMetrics(): Promise { + const all = await this.getAllEvents(); + const eventsByType: Record = {}; + const aggregatesByType: Record = {}; + let lastEventAt: string | undefined; + + for (const event of all) { + eventsByType[event.eventType] = (eventsByType[event.eventType] ?? 0) + 1; + aggregatesByType[event.aggregateType] = (aggregatesByType[event.aggregateType] ?? 0) + 1; + if (!lastEventAt || event.occurredAt > lastEventAt) { + lastEventAt = event.occurredAt; + } + } + + return { + totalEvents: all.length, + eventsByType, + aggregatesByType, + lastEventAt, + }; + } + + /** Clear all events (tests only). */ + async clear(): Promise { + if (this.useDb && this.db.transactionEvent) { + await this.db.transactionEvent.deleteMany({ where: {} }); + } else { + await this.memory.clear(); + } + } + + private toStoredEvent(row: PersistedEvent): StoredEvent { + return { + id: row.id, + aggregateType: row.aggregateType, + aggregateId: row.aggregateId, + eventType: row.eventType, + payload: row.payload, + version: row.version, + occurredAt: row.occurredAt, + sequenceNumber: row.sequenceNumber, + correlationId: row.correlationId ?? undefined, + causationId: row.causationId ?? undefined, + userId: row.userId ?? undefined, + metadata: row.metadata ?? undefined, + }; + } +} + +/** Singleton accessor. */ +let _eventStore: EventStore | null = null; + +export function getEventStore(): EventStore { + if (!_eventStore) _eventStore = new EventStore(); + return _eventStore; +} + +export function resetEventStoreForTesting(): void { + _eventStore = null; +} diff --git a/backend/src/eventsourcing/projection.ts b/backend/src/eventsourcing/projection.ts new file mode 100644 index 00000000..7e8cd5f1 --- /dev/null +++ b/backend/src/eventsourcing/projection.ts @@ -0,0 +1,87 @@ +/** + * projection.ts — Issue #946 + * + * Projection base class for building read models from event streams. + * + * Projections subscribe to event types and maintain a denormalized read model. + * They are eventually consistent: events are applied asynchronously after the + * aggregate commits them. + */ + +import type { StoredEvent, EventHandler, ProjectionSubscription } from './types.js'; +import { getEventStore } from './event-store.js'; + +/** Base class for projections. Subclasses implement `subscribe()` and event + * handler methods keyed by event type. */ +export abstract class Projection { + protected abstract readonly name: string; + protected abstract readonly eventTypes: string[]; + + /** Return the subscription descriptor used by the projection manager. */ + getSubscription(): ProjectionSubscription { + return { + projectionName: this.name, + eventTypes: this.eventTypes, + handler: async (event: StoredEvent) => { + const handlerName = `on${event.eventType.charAt(0).toUpperCase()}${event.eventType.slice(1)}`; + const handler = (this as unknown as Record)[handlerName]; + if (typeof handler === 'function') { + await handler.call(this, event); + } + }, + }; + } + + /** Projected read model accessor (subclasses override). */ + abstract getReadModel(aggregateId: string): unknown; +} + +/** Manages projections and replays events into them. */ +export class ProjectionManager { + private readonly projections = new Map(); + private readonly store = getEventStore(); + private running = false; + + register(projection: Projection): void { + this.projections.set(projection.name, projection); + } + + /** Replay all historical events into every registered projection. */ + async replay(): Promise { + const events = await this.store.getAllEvents(); + for (const event of events) { + for (const projection of this.projections.values()) { + const sub = projection.getSubscription(); + if (sub.eventTypes.includes(event.eventType)) { + await sub.handler(event); + } + } + } + } + + /** Project a single new event into all interested projections. */ + async project(event: StoredEvent): Promise { + for (const projection of this.projections.values()) { + const sub = projection.getSubscription(); + if (sub.eventTypes.includes(event.eventType)) { + await sub.handler(event); + } + } + } + + getProjectionNames(): string[] { + return Array.from(this.projections.keys()); + } +} + +/** Singleton projection manager. */ +let _manager: ProjectionManager | null = null; + +export function getProjectionManager(): ProjectionManager { + if (!_manager) _manager = new ProjectionManager(); + return _manager; +} + +export function resetProjectionManagerForTesting(): void { + _manager = null; +} diff --git a/backend/src/eventsourcing/transaction-aggregate.ts b/backend/src/eventsourcing/transaction-aggregate.ts new file mode 100644 index 00000000..8ad280aa --- /dev/null +++ b/backend/src/eventsourcing/transaction-aggregate.ts @@ -0,0 +1,240 @@ +/** + * transaction-aggregate.ts — Issue #946 + * + * Transaction aggregate root implementing event sourcing for payment + * transactions. State transitions are expressed as domain events: + * + * TransactionCreated → PaymentAuthorized → PaymentCaptured + * ↳ PaymentFailed + * ↳ TransactionRefunded + * + * Each command validates invariants before emitting its event. + */ + +import { ok, err, type Result } from '../../lib/result.js'; +import { AggregateRoot, type AggregateState } from './aggregate.js'; +import type { StoredEvent } from './types.js'; + +export type TransactionStatus = 'pending' | 'authorized' | 'captured' | 'failed' | 'refunded' | 'cancelled'; + +export interface TransactionState extends AggregateState { + status: TransactionStatus; + amount: number; + currency: string; + toAddress: string; + fromAddress?: string; + network: string; + providerId?: string; + txHash?: string; + failureReason?: string; + refundedAmount: number; + createdAt: string; + updatedAt: string; +} + +const initialState = (overrides: Partial = {}): TransactionState => ({ + status: 'pending', + amount: 0, + currency: 'XLM', + toAddress: '', + network: 'stellar', + refundedAmount: 0, + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + ...overrides, +}); + +export class TransactionAggregate extends AggregateRoot { + protected readonly aggregateType = 'transaction'; + protected readonly id: string; + + constructor(id: string, initial?: TransactionState) { + super(initial ?? initialState({ id })); + this.id = id; + } + + protected apply(event: StoredEvent): void { + switch (event.eventType) { + case 'TransactionCreated': { + const p = event.payload as Partial; + this.state = { + ...this.state, + status: 'pending', + amount: p.amount ?? 0, + currency: p.currency ?? 'XLM', + toAddress: p.toAddress ?? '', + fromAddress: p.fromAddress, + network: p.network ?? 'stellar', + createdAt: event.occurredAt, + updatedAt: event.occurredAt, + }; + break; + } + case 'PaymentAuthorized': { + const p = event.payload as { providerId?: string; txHash?: string }; + this.state = { + ...this.state, + status: 'authorized', + providerId: p.providerId ?? this.state.providerId, + txHash: p.txHash ?? this.state.txHash, + updatedAt: event.occurredAt, + }; + break; + } + case 'PaymentCaptured': { + this.state = { ...this.state, status: 'captured', updatedAt: event.occurredAt }; + break; + } + case 'PaymentFailed': { + const p = event.payload as { reason?: string }; + this.state = { + ...this.state, + status: 'failed', + failureReason: p.reason, + updatedAt: event.occurredAt, + }; + break; + } + case 'TransactionRefunded': { + const p = event.payload as { amount?: number }; + const refunded = (this.state.refundedAmount ?? 0) + (p.amount ?? 0); + this.state = { + ...this.state, + status: refunded >= this.state.amount ? 'refunded' : 'partially_refunded', + refundedAmount: refunded, + updatedAt: event.occurredAt, + }; + break; + } + case 'TransactionCancelled': { + this.state = { ...this.state, status: 'cancelled', updatedAt: event.occurredAt }; + break; + } + } + } + + /** Create a new transaction. Idempotent: rejected if already created. */ + async createTransaction( + input: { amount: number; currency: string; toAddress: string; fromAddress?: string; network?: string; userId?: string }, + ): Promise> { + if (this.version > 0) { + return err({ + code: 'TRANSACTION_ALREADY_EXISTS', + message: `Transaction ${this.id} already exists`, + statusCode: 409, + }); + } + if (!Number.isFinite(input.amount) || input.amount <= 0) { + return err({ + code: 'INVALID_AMOUNT', + message: 'Transaction amount must be a finite number greater than zero', + statusCode: 400, + }); + } + + await this.commit('TransactionCreated', { + amount: input.amount, + currency: input.currency, + toAddress: input.toAddress, + fromAddress: input.fromAddress, + network: input.network ?? 'stellar', + userId: input.userId, + }); + + return ok(this.state); + } + + /** Authorize a payment against a downstream provider. */ + async authorizePayment(input: { providerId: string; txHash?: string; userId?: string }): Promise> { + if (this.state.status !== 'pending') { + return err({ + code: 'INVALID_TRANSITION', + message: `Cannot authorize transaction in state ${this.state.status}`, + statusCode: 409, + }); + } + + await this.commit('PaymentAuthorized', { + providerId: input.providerId, + txHash: input.txHash, + userId: input.userId, + }); + + return ok(this.state); + } + + /** Capture an authorized payment. */ + async capturePayment(userId?: string): Promise> { + if (this.state.status !== 'authorized') { + return err({ + code: 'INVALID_TRANSITION', + message: `Cannot capture transaction in state ${this.state.status}`, + statusCode: 409, + }); + } + + await this.commit('PaymentCaptured', {}, { userId }); + return ok(this.state); + } + + /** Mark a payment as failed. */ + async failPayment(reason: string, userId?: string): Promise> { + if (this.state.status === 'captured' || this.state.status === 'refunded') { + return err({ + code: 'INVALID_TRANSITION', + message: `Cannot fail transaction in state ${this.state.status}`, + statusCode: 409, + }); + } + + await this.commit('PaymentFailed', { reason }, { userId }); + return ok(this.state); + } + + /** Refund part or all of a captured transaction. */ + async refund(input: { amount: number; userId?: string }): Promise> { + if (this.state.status !== 'captured' && this.state.status !== 'authorized') { + return err({ + code: 'INVALID_TRANSITION', + message: `Cannot refund transaction in state ${this.state.status}`, + statusCode: 409, + }); + } + if (!Number.isFinite(input.amount) || input.amount <= 0) { + return err({ + code: 'INVALID_AMOUNT', + message: 'Refund amount must be greater than zero', + statusCode: 400, + }); + } + if (input.amount + this.state.refundedAmount > this.state.amount) { + return err({ + code: 'REFUND_EXCEEDS_AMOUNT', + message: 'Refund amount exceeds the remaining transaction amount', + statusCode: 400, + }); + } + + await this.commit('TransactionRefunded', { amount: input.amount }, { userId }); + return ok(this.state); + } + + /** Cancel a pending transaction. */ + async cancel(userId?: string): Promise> { + if (this.state.status !== 'pending') { + return err({ + code: 'INVALID_TRANSITION', + message: `Cannot cancel transaction in state ${this.state.status}`, + statusCode: 409, + }); + } + + await this.commit('TransactionCancelled', {}, { userId }); + return ok(this.state); + } + + /** Get current state snapshot. */ + getState(): TransactionState { + return { ...this.state }; + } +} diff --git a/backend/src/eventsourcing/transaction-projector.ts b/backend/src/eventsourcing/transaction-projector.ts new file mode 100644 index 00000000..30d70594 --- /dev/null +++ b/backend/src/eventsourcing/transaction-projector.ts @@ -0,0 +1,127 @@ +/** + * transaction-projector.ts — Issue #946 + * + * Read-model projection for transactions. Maintains a denormalized view of + * transaction state keyed by transaction id, updated from domain events. + * + * In production this would write to a dedicated read table or search index. + * Here it maintains an in-memory map for demonstration and testing. + */ + +import type { StoredEvent } from './types.js'; +import { Projection } from './projection.js'; + +interface TransactionReadModel { + id: string; + status: string; + amount: number; + currency: string; + toAddress: string; + providerId?: string; + txHash?: string; + refundedAmount: number; + updatedAt: string; +} + +export class TransactionProjector extends Projection { + protected readonly name = 'transaction-read-model'; + protected readonly eventTypes = [ + 'TransactionCreated', + 'PaymentAuthorized', + 'PaymentCaptured', + 'PaymentFailed', + 'TransactionRefunded', + 'TransactionCancelled', + ]; + + private readonly models = new Map(); + + getReadModel(aggregateId: string): TransactionReadModel | undefined { + return this.models.get(aggregateId); + } + + getAll(): TransactionReadModel[] { + return Array.from(this.models.values()); + } + + private onTransactionCreated(event: StoredEvent): void { + const p = event.payload as TransactionReadModel & { fromAddress?: string; network?: string }; + this.models.set(event.aggregateId, { + id: event.aggregateId, + status: 'pending', + amount: p.amount ?? 0, + currency: p.currency ?? 'XLM', + toAddress: p.toAddress ?? '', + refundedAmount: 0, + updatedAt: event.occurredAt, + }); + } + + private onPaymentAuthorized(event: StoredEvent): void { + const model = this.models.get(event.aggregateId); + if (!model) return; + const p = event.payload as { providerId?: string; txHash?: string }; + this.models.set(event.aggregateId, { + ...model, + status: 'authorized', + providerId: p.providerId, + txHash: p.txHash, + updatedAt: event.occurredAt, + }); + } + + private onPaymentCaptured(event: StoredEvent): void { + const model = this.models.get(event.aggregateId); + if (!model) return; + this.models.set(event.aggregateId, { + ...model, + status: 'captured', + updatedAt: event.occurredAt, + }); + } + + private onPaymentFailed(event: StoredEvent): void { + const model = this.models.get(event.aggregateId); + if (!model) return; + this.models.set(event.aggregateId, { + ...model, + status: 'failed', + updatedAt: event.occurredAt, + }); + } + + private onTransactionRefunded(event: StoredEvent): void { + const model = this.models.get(event.aggregateId); + if (!model) return; + const p = event.payload as { amount?: number }; + const refunded = (model.refundedAmount ?? 0) + (p.amount ?? 0); + this.models.set(event.aggregateId, { + ...model, + status: refunded >= model.amount ? 'refunded' : model.status, + refundedAmount: refunded, + updatedAt: event.occurredAt, + }); + } + + private onTransactionCancelled(event: StoredEvent): void { + const model = this.models.get(event.aggregateId); + if (!model) return; + this.models.set(event.aggregateId, { + ...model, + status: 'cancelled', + updatedAt: event.occurredAt, + }); + } +} + +/** Singleton accessor. */ +let _projector: TransactionProjector | null = null; + +export function getTransactionProjector(): TransactionProjector { + if (!_projector) _projector = new TransactionProjector(); + return _projector; +} + +export function resetTransactionProjectorForTesting(): void { + _projector = null; +} diff --git a/backend/src/eventsourcing/types.ts b/backend/src/eventsourcing/types.ts new file mode 100644 index 00000000..338a799d --- /dev/null +++ b/backend/src/eventsourcing/types.ts @@ -0,0 +1,66 @@ +/** + * types.ts — Issue #946 + * + * Core types for the event sourcing architecture used by transactions. + */ + +import type { Result } from '../../lib/result.js'; + +/** An immutable fact that something happened in the domain. */ +export interface StoredEvent { + id: string; + /** e.g. "transaction", "payment", "refund" */ + aggregateType: string; + /** The aggregate instance id (e.g. a transaction id) */ + aggregateId: string; + /** e.g. "TransactionCreated", "PaymentAuthorized", "TransactionCompleted" */ + eventType: string; + /** The event payload (domain data). */ + payload: T; + /** Monotonically increasing version within this aggregate stream. */ + version: number; + /** When the event occurred. */ + occurredAt: string; + /** Causal position in the global event log. */ + sequenceNumber: number; + /** Optional correlation id linking related events (e.g. a payment to its transaction). */ + correlationId?: string; + /** Optional causation id (the event that caused this one). */ + causationId?: string; + /** User/system that initiated the event. */ + userId?: string; + /** Additional metadata (tenant, trace id, etc.). */ + metadata?: Record; +} + +/** A snapshot of aggregate state used for faster rehydration. */ +export interface AggregateSnapshot { + aggregateId: string; + aggregateType: string; + version: number; + state: T; + createdAt: string; +} + +/** Command result returned by aggregate methods. */ +export type CommandResult = Result; + +/** An event handler used by projections. */ +export interface EventHandler { + (event: StoredEvent): void | Promise; +} + +/** Projection subscription entry. */ +export interface ProjectionSubscription { + projectionName: string; + eventTypes: string[]; + handler: EventHandler; +} + +/** Metrics for the event store. */ +export interface EventStoreMetrics { + totalEvents: number; + eventsByType: Record; + aggregatesByType: Record; + lastEventAt?: string; +}