What this module covers: A blockchain indexer processing 50,000 events/second will accumulate complex business rules over time: transaction validity invariants, account balance constraints, settlement reconciliation logic, fraud detection rules. Without architectural discipline, these rules scatter across route handlers, database queries, and event listeners — untestable, fragile, and impossible to reason about under load. This module covers Domain-Driven Design applied to state-heavy systems, Clean Architecture for transport-agnostic domain logic, and CQRS for systems where the read model and write model need to evolve independently.
Why Architecture Matters More at Scale
At low throughput (< 1K events/sec), architectural shortcuts are invisible. Validation in route handlers, business logic in SQL queries, domain concepts scattered across layers — these are maintainability problems, not performance problems.
At high throughput, architectural shortcuts become performance problems:
javascript
Problems:
Untestable: you need a running HTTP server and database to test the business logic
Uncacheable: the business rules are coupled to the SQL and HTTP transport
Unreusable: the same logic cannot be reused by a gRPC endpoint or a Kafka consumer
Unrefactorable: changing the validation requires touching the route handler
Domain-Driven Design Core Concepts
DDD provides vocabulary and patterns for modeling complex business domains in code. For a payment system, the key concepts are:
Entities
Objects with identity that persists over time. An account is an entity — it has an ID, and its state changes.
typescript
Why this matters for high-throughput systems:
The invariants are in the domain object and enforced before any I/O happens. At 50K payments/second, you want to reject invalid payments with zero database roundtrips. The Account.debit() method checks balance without touching the database — if it throws, the payment is rejected before any SQL runs.
Value Objects
Immutable objects without identity. An Amount is a value object — two amounts of 100 USD are identical regardless of which "instance" they are.
typescript
Aggregates: The Consistency Boundary
An aggregate is a cluster of entities and value objects that must be changed together atomically. The aggregate root is the entry point — external code can only access the aggregate through the root.
An aggregate is the blast radius of one transaction — everything inside it goes up together or not at all; everything outside only ever hears about it afterward, through a message slipped under the door (a domain event). That last part matters: the standard DDD invariant is that a single transaction should modify exactly one aggregate. Two aggregates never get mutated in the same in-process call — the second one finds out via an event, on its own transaction, after the first has already committed.
The naive version of a Payment breaks this rule directly — it reaches into two separate account aggregates from inside a single method call:
typescript
The fix — one aggregate per transaction, coordinated by a domain event:Payment.process() (or, better, the application service driving it) modifies exactly one side — the sender's debit — as its own transaction, and emits MoneyDebited. A process manager (a saga) reacts to that event and issues the credit to the recipient as a separate transaction. If the credit fails, the process manager issues a compensating action (reversing the debit) rather than the whole operation silently rolling back two aggregates it never should have touched together in the first place.
typescript
typescript
Deadlock avoidance — consistent lock ordering: even with the saga restructuring above, both the debit step and the compensation step take a row lock on a single account row per transaction — that's the entire point of splitting them apart. But some real ledger systems intentionally lock two account rows in one transaction for a synchronous transfer (skipping the saga for latency reasons on a trusted internal path). If you do that, you must acquire the locks in a consistent global order — never "sender then recipient," always e.g. "lower account ID then higher account ID" — or two concurrent transfers in opposite directions (A→B and B→A) will each hold one lock and block waiting for the other, and the database's deadlock detector kills one of them.
typescript
Production incident — deadlocks under peak UPI festival load: a payment path that intentionally locked both accounts in one transaction (for latency reasons, bypassing the saga on a trusted internal transfer path) originally locked sender first, then recipient — whatever order they arrived in the request. Under normal load this was fine. During a festival peak, concurrent A→B and B→A transfers between the same pair of popular merchant accounts became common, and Postgres began throwing "deadlock detected" errors on a meaningful fraction of transfers — transaction A held a lock on account A waiting for account B, while transaction B (the reverse transfer) held a lock on account B waiting for account A, and neither could proceed until Postgres killed one. The fix was exactly the pattern above: always lock accounts in ascending ID order, regardless of which one is the "sender" in that particular request. Once shipped, the deadlocks disappeared entirely — with a consistent lock order, two transactions can never form a cycle waiting on each other.
Clean Architecture: The Dependency Rule
Clean Architecture enforces that dependencies only point inward — from infrastructure to application to domain. The domain never knows about HTTP, Kafka, or PostgreSQL.
text
Ports and Adapters (Hexagonal Architecture)
typescript
typescript
Why the unconditional UPDATE was a lost-update bug: an UPDATE accounts SET balance = $1 ... WHERE id = $3 with no version check always succeeds and always overwrites whatever is currently in the row — it has no way to detect that another transaction read the same row, computed a different new balance, and already wrote it. Two concurrent debits against the same account (e.g. one from ProcessPaymentUseCase, one from a compensating action in the saga above) can both read balance 1000, both compute their own decremented value in memory, and whichever UPDATE runs last simply clobbers the other — one of the two debits is silently lost, and the balance ends up wrong by exactly that amount. Adding version = $4 to the WHERE clause and requiring rowCount > 0 turns this into a compare-and-swap: the second writer's UPDATE matches zero rows (the version it read is now stale), fails loudly with ConcurrentModificationError, and the caller can retry by re-reading the account and re-applying its operation — instead of silently losing an update.
The transport-agnostic domain: the same ProcessPaymentUseCase works whether triggered by:
typescript
The domain and application layers are untouched. Only the infrastructure adapter changes.
Dependency Injection: Wiring the Layers
typescript
Swap PostgreSQL for an in-memory repository for unit tests without touching any business logic:
typescript
Tests run in milliseconds. No database. No network. Business logic tested in isolation.
CQRS: Separate Write and Read Models
Command Query Responsibility Segregation (CQRS) separates the model that handles writes (commands) from the model that handles reads (queries).
For a payment ledger:
Write model: normalized, ACID-consistent. Enforces invariants. Optimized for integrity.
Read model: denormalized, eventually consistent. Pre-computed aggregations. Optimized for query speed.
typescript
The read model is updated by an event consumer that processes domain events from Kafka:
typescript
CQRS benefits for high-throughput systems:
Write and read databases can be scaled independently
Read queries never compete with write transactions
The read model can be optimized for specific query patterns (denormalized, indexed differently)
If the read model becomes stale, you can rebuild it by replaying events from Kafka
Anti-Corruption Layer for Blockchain RPC
When your indexer calls a third-party blockchain RPC node (Ethereum's eth_getBlockByNumber, Supra's block API), the response structure is the external system's format. Letting that format leak into your domain creates coupling — if the RPC response format changes, your domain objects break.
typescript
The domain Block object uses your types (Buffer for hashes, BigInt for amounts) and a correctly named gasLimit — not a gasUsed that was never actually measured. If a workflow genuinely needs gas consumption (for fee reconciliation, say), fetch it explicitly via getGasUsed() against the receipt, and keep it as a separate, optional field populated after execution — never conflate it with the limit taken from the block. If Ethereum changes their API response format, you update translateEthBlock — nowhere else.
Production Incident: Domain Logic in a SQL Query
Context: A banking ledger service. Balance calculations were done in SQL via a stored procedure called by the application:
sql
What happened:
A regulatory requirement changed: transactions pending for more than 48 hours must be included in the balance calculation as "provisional debits." The business rule changed from:
Balance = sum of settled transactions
To:
Balance = sum of settled transactions + sum of pending debits older than 48 hours
The stored procedure was updated. But the update missed that pending credits older than 48 hours should also be included (oversight). The bug was in production for 11 days before a reconciliation audit caught it. 847 accounts had incorrect balance calculations.
Root cause: the business rule was in SQL, not in domain code. SQL is not testable with unit tests. The domain rule was invisible to developers who tested in isolation.
The fix — move the rule to the domain:
typescript
The test caught the exact bug — credits and debits treated consistently — before any code reached production.
Summary
Concept
Key Takeaway
Entity
Object with identity that changes over time. Enforces its own invariants.
Value Object
Immutable, identity-less. Money(5000, 'INR') is always equal to another Money(5000, 'INR').
Aggregate
Cluster of entities with one root. Changed atomically. Emits domain events.
Invariant
Rule the domain enforces. Never in SQL. Always in domain code. Testable without I/O.
Translate external formats at the boundary. External changes never reach domain objects.
Domain events
Emitted by aggregates after state changes. Published to Kafka by the infrastructure layer.
One aggregate per transaction
Payment.process() must not mutate sender and recipient in one call. Debit one, emit MoneyDebited, credit the other via a saga.
Consistent lock ordering
If a transaction must lock two account rows, always lock them in the same global order (e.g. by ID) to prevent deadlocks.
Optimistic concurrency
Unconditional UPDATE is a lost-update bug. Add a version column and a WHERE version = $n compare-and-swap.
The architecture is clean. Module 12 covers how to keep it healthy under production load — CPU profiling, flame graph reading, ELU monitoring, and the full clinic.js diagnostic workflow for Node.js services under high-throughput stress.
// BAD: business logic in the route handlerapp.post('/api/v2/payments',async(req, res)=>{const{ senderId, recipientId, amount }= req.body;// Business logic in HTTP handler:const sender =await db.query('SELECT balance FROM accounts WHERE id = $1',[senderId]);if(!sender.rows[0])return res.status(404).json({error:'Sender not found'});if(sender.rows[0].balance< amount)return res.status(400).json({error:'Insufficient funds'});// More business logic:const fraud =awaitcheckFraud(senderId, amount);if(fraud.risk>0.8)return res.status(403).json({error:'High risk transaction'});// More business logic:await db.query('BEGIN');await db.query('UPDATE accounts SET balance = balance - $1 WHERE id = $2',[amount, senderId]);await db.query('UPDATE accounts SET balance = balance + $1 WHERE id = $2',[amount, recipientId]);await db.query('INSERT INTO transactions VALUES ...',[...]);await db.query('COMMIT'); res.json({status:'accepted'});});
// modules/accounts/domain/Account.tsexportclassAccount{ #id:string; #balance: bigint; #status:'active'|'frozen'|'closed'; #version:number;// optimistic concurrency token — see the repository below #events: DomainEvent[]=[];privateconstructor(id:string, balance: bigint, status:'active'|'frozen'|'closed', version:number){this.#id = id;this.#balance = balance;this.#status = status;this.#version = version;}staticcreate(id:string, initialBalance: bigint): Account {const account =newAccount(id, initialBalance,'active',0); account.#events.push(newAccountCreatedEvent(id, initialBalance));return account;}staticreconstitute(id:string, balance: bigint, status:'active'|'frozen'|'closed', version:number): Account {returnnewAccount(id, balance, status, version);}getversion(){returnthis.#version;}debit(amount: bigint, reference:string):void{// Invariant enforcement — IN the domain object, not in application codeif(this.#status !=='active'){thrownewAccountNotActiveError(this.#id,this.#status);}if(amount <=0n){thrownewInvalidAmountError('Debit amount must be positive');}if(this.#balance < amount){thrownewInsufficientFundsError(this.#id,this.#balance, amount);}this.#balance -= amount;this.#events.push(newAccountDebitedEvent(this.#id, amount, reference));}credit(amount: bigint, reference:string):void{if(this.#status ==='closed'){thrownewAccountClosedError(this.#id);}if(amount <=0n)thrownewInvalidAmountError('Credit amount must be positive');this.#balance += amount;this.#events.push(newAccountCreditedEvent(this.#id, amount, reference));}freeze():void{if(this.#status !=='active')thrownewAccountNotActiveError(this.#id,this.#status);this.#status ='frozen';this.#events.push(newAccountFrozenEvent(this.#id));}getid(){returnthis.#id;}getbalance(){returnthis.#balance;}getstatus(){returnthis.#status;}getevents(){return[...this.#events];}clearEvents(){this.#events =[];}}
// WRONG: Payment.process() mutates two aggregate roots in one call.// sender and recipient are separate Account aggregates — this couples// their consistency boundaries and, done naively against a real database,// requires locking both rows in a single transaction (see below).process():void{if(this.#status !=='pending')thrownewInvalidStateTransitionError('process',this.#status);this.#sender.debit(this.#amount.amount,this.#id);// aggregate #1this.#recipient.credit(this.#amount.amount,this.#id);// aggregate #2 — should not happen herethis.#status ='completed';this.#events.push(newPaymentCompletedEvent(this.#id));}
// The Payment aggregate now only drives ONE aggregate's mutation directly.exportclassPayment{ #id:string; #senderId:string; #recipientId:string; #amount: Money; #status:'pending'|'processing'|'completed'|'failed'; #events: DomainEvent[]=[];staticinitiate(sender: Account, recipient: Account, amount: Money): Payment {if(sender.id === recipient.id)thrownewSelfTransferError();if(sender.status !=='active')thrownewAccountNotActiveError(sender.id, sender.status);if(sender.balance < amount.amount)thrownewInsufficientFundsError(sender.id, sender.balance, amount.amount);const payment =newPayment(generateId(), sender.id, recipient.id, amount,'pending'); payment.#events.push(newPaymentInitiatedEvent(payment.#id, sender.id, recipient.id, amount));return payment;}// Modifies ONLY the sender aggregate. The recipient is updated later,// by a separate transaction, in reaction to the MoneyDebited event below.debitSender(sender: Account):void{if(this.#status !=='pending')thrownewInvalidStateTransitionError('debitSender',this.#status); sender.debit(this.#amount.amount,this.#id);this.#status ='processing';this.#events.push(newMoneyDebitedEvent(this.#id,this.#senderId,this.#recipientId,this.#amount));}complete():void{this.#status ='completed';this.#events.push(newPaymentCompletedEvent(this.#id));}fail(reason:string):void{this.#status ='failed';this.#events.push(newPaymentFailedEvent(this.#id, reason));}}
// Process manager (saga): reacts to MoneyDebited in its OWN transaction,// crediting the recipient aggregate separately from the sender's debit.classPaymentSagaHandler{constructor(private accounts: AccountRepository,private events: DomainEventPublisher){}// Triggered by MoneyDebited — runs as an independent transaction against// the recipient aggregate only. Never touches the sender in this step.asynconMoneyDebited(event: MoneyDebitedEvent):Promise<void>{try{const recipient =awaitthis.accounts.findById(event.recipientId); recipient.credit(event.amount.amount, event.paymentId);awaitthis.accounts.save(recipient);awaitthis.events.publish(newMoneyCreditedEvent(event.paymentId, event.recipientId, event.amount));}catch(err){// Compensating action: reverse the sender's debit in its own// transaction, rather than pretending the two aggregates ever// shared a rollback boundary.awaitthis.events.publish(newPaymentCompensationRequiredEvent(event.paymentId, event.senderId, event.amount));}}asynconPaymentCompensationRequired(event: PaymentCompensationRequiredEvent):Promise<void>{const sender =awaitthis.accounts.findById(event.senderId); sender.credit(event.amount.amount,`compensation:${event.paymentId}`);// reverse the debitawaitthis.accounts.save(sender);}}
// If two account rows must be locked in ONE transaction, always acquire// them in a consistent order — sorted by ID, never by "sender then recipient"asyncfunctionlockAccountsInOrder(client: PoolClient, accountIdA:string, accountIdB:string){const[first, second]=[accountIdA, accountIdB].sort();// deterministic, not request-dependentconst firstRow =await client.query('SELECT * FROM accounts WHERE id = $1 FOR UPDATE',[first]);const secondRow =await client.query('SELECT * FROM accounts WHERE id = $1 FOR UPDATE',[second]);return{ firstRow, secondRow };}
// Domain layer: defines the PORT (interface)// ports/AccountRepository.tsexportinterfaceAccountRepository{findById(id:string):Promise<Account |null>;save(account: Account):Promise<void>;}// Application layer: uses the PORT, knows nothing about PostgreSQL// use-cases/ProcessPayment.ts//// Note: this use case now touches only ONE aggregate (the sender's Account).// The recipient's credit happens later, in PaymentSagaHandler.onMoneyDebited,// reacting to the MoneyDebited event this use case publishes below — see// "Aggregates: The Consistency Boundary" for why the two are never mutated// in the same call.exportclassProcessPaymentUseCase{constructor(private accounts: AccountRepository,// injected via DIprivate events: DomainEventPublisher,){}asyncexecute(senderId:string, recipientId:string, amount: Money):Promise<Payment>{const[sender, recipient]=awaitPromise.all([this.accounts.findById(senderId),this.accounts.findById(recipientId),]);if(!sender)thrownewAccountNotFoundError(senderId);if(!recipient)thrownewAccountNotFoundError(recipientId);const payment = Payment.initiate(sender, recipient, amount); payment.debitSender(sender);// mutates the sender aggregate onlyawaitthis.accounts.save(sender);// one aggregate, one transaction// Publish domain events — MoneyDebited is among them, and drives the// recipient's credit via the saga handler, in its own transactionfor(const event of payment.events){awaitthis.events.publish(event);}return payment;}}
// Infrastructure layer: implements the PORT for PostgreSQL (the ADAPTER)// adapters/PostgresAccountRepository.tsimport{ Account }from'../../domain/Account';import{ AccountRepository }from'../../ports/AccountRepository';exportclassPostgresAccountRepositoryimplementsAccountRepository{constructor(private pool: Pool){}asyncfindById(id:string):Promise<Account |null>{const{ rows }=awaitthis.pool.query('SELECT id, balance, status, version FROM accounts WHERE id = $1',[id]);if(!rows[0])returnnull;// The version read here travels with the in-memory Account and is// compared-and-swapped on save() below — see "Optimistic Concurrency// Control" for why an unconditional UPDATE isn't safe on its own.return Account.reconstitute(rows[0].id,BigInt(rows[0].balance), rows[0].status, rows[0].version);}asyncsave(account: Account):Promise<void>{const{ rowCount }=awaitthis.pool.query(`UPDATE accounts
SET balance = $1, status = $2, version = version + 1
WHERE id = $3 AND version = $4`,[account.balance.toString(), account.status, account.id, account.version]);if(rowCount ===0){// Someone else updated this row (and bumped its version) between our// findById() and this save() — our in-memory Account was stale.thrownewConcurrentModificationError(account.id);}}}
// composition-root/container.ts — wire everything together at startupimport{ Pool }from'pg';import{ Kafka }from'kafkajs';import{ PostgresAccountRepository }from'../adapters/PostgresAccountRepository';import{ KafkaEventPublisher }from'../adapters/KafkaEventPublisher';import{ ProcessPaymentUseCase }from'../use-cases/ProcessPaymentUseCase';exportfunctionbuildContainer(){const pool =newPool({ connectionString: process.env.DATABASE_URL});const kafka =newKafka({ brokers: process.env.KAFKA_BROKERS!.split(',')});const producer = kafka.producer();const accountRepo =newPostgresAccountRepository(pool);const eventPublisher =newKafkaEventPublisher(producer);const processPayment =newProcessPaymentUseCase(accountRepo, eventPublisher);return{ processPayment, pool, producer };}
// tests/use-cases/ProcessPayment.test.tsimport{ InMemoryAccountRepository }from'../test-doubles/InMemoryAccountRepository';import{ SpyEventPublisher }from'../test-doubles/SpyEventPublisher';describe('ProcessPaymentUseCase',()=>{const accounts =newInMemoryAccountRepository();const events =newSpyEventPublisher();const useCase =newProcessPaymentUseCase(accounts, events);beforeEach(()=>{ accounts.seed(Account.reconstitute('sender-1',10000n,'active',0)); accounts.seed(Account.reconstitute('recipient-1',0n,'active',0));});it('debits the sender and publishes MoneyDebited — recipient is untouched here',async()=>{// execute() only ever mutates the sender aggregate, per-transaction.// The recipient's credit is a separate concern, handled by// PaymentSagaHandler.onMoneyDebited reacting to the published event —// see "Aggregates: The Consistency Boundary" for why.await useCase.execute('sender-1','recipient-1',newMoney(5000n,'INR'));const sender =await accounts.findById('sender-1');const recipient =await accounts.findById('recipient-1');expect(sender!.balance).toBe(5000n);expect(recipient!.balance).toBe(0n);// not yet credited — that's the saga's jobexpect(events.published).toContainEqual( expect.objectContaining({ type:'MoneyDebited', accountId:'sender-1'}));});it('rejects payment when balance is insufficient',async()=>{awaitexpect( useCase.execute('sender-1','recipient-1',newMoney(20000n,'INR'))).rejects.toThrow(InsufficientFundsError);});});describe('PaymentSagaHandler',()=>{it('credits the recipient in its own transaction, reacting to MoneyDebited',async()=>{const accounts =newInMemoryAccountRepository(); accounts.seed(Account.reconstitute('recipient-1',0n,'active',0));const events =newSpyEventPublisher();const saga =newPaymentSagaHandler(accounts, events);await saga.onMoneyDebited({ paymentId:'pay-1', senderId:'sender-1', recipientId:'recipient-1', amount:newMoney(5000n,'INR'),});const recipient =await accounts.findById('recipient-1');expect(recipient!.balance).toBe(5000n);});});
// Write side: command handler (uses the rich domain model)classDebitAccountCommandHandler{asynchandle(cmd: DebitAccountCommand):Promise<void>{const account =awaitthis.accountRepo.findById(cmd.accountId); account.debit(cmd.amount, cmd.reference);// invariant enforcementawaitthis.accountRepo.save(account);awaitthis.eventBus.publish(account.events);}}// Read side: query handler (uses simple, fast projections)classGetAccountSummaryQueryHandler{asynchandle(query: GetAccountSummaryQuery):Promise<AccountSummary>{// Read from pre-computed projection table — no domain model neededconst{ rows }=awaitthis.readPool.query(`SELECT
account_id,
balance,
total_debited_30d,
total_credited_30d,
transaction_count_30d,
last_transaction_at
FROM account_summaries
WHERE account_id = $1`,[query.accountId]);return rows[0];}}
// Projection builder: maintains the read modelconsumer.run({eachMessage:async({ message })=>{const event =JSON.parse(message.value.toString());if(event.type ==='ACCOUNT_DEBITED'){await readPool.query(`INSERT INTO account_summaries (account_id, balance, total_debited_30d, transaction_count_30d, last_transaction_at)
VALUES ($1, -$2, $2, 1, NOW())
ON CONFLICT (account_id) DO UPDATE SET
balance = account_summaries.balance - $2,
total_debited_30d = account_summaries.total_debited_30d + $2,
transaction_count_30d = account_summaries.transaction_count_30d + 1,
last_transaction_at = NOW()`,[event.accountId,BigInt(event.amount)]);}}});
// External world: raw Ethereum RPC response from eth_getBlockByNumberinterfaceEthRpcBlock{number:string;// hex string: "0x11A3E70" hash:string;// 0x-prefixed hex transactions:Array<{ hash:string; from:string; to:string|null; value:string;// hex string in wei gas:string;// the gas LIMIT the sender authorized — NOT what was consumed gasPrice:string; input:string;}>;}// Anti-corruption layer: translate to your domain modelfunctiontranslateEthBlock(raw: EthRpcBlock): Block {returnnewBlock({ height:parseInt(raw.number,16),// hex → number hash: Buffer.from(raw.hash.slice(2),'hex'),// hex string → Buffer transactions: raw.transactions.map(tx =>newTransaction({ hash: Buffer.from(tx.hash.slice(2),'hex'), sender: tx.from, recipient: tx.to ??null, amount:BigInt(tx.value),// hex string → BigInt// `tx.gas` from eth_getBlockByNumber is the gas LIMIT — the cap the// sender was willing to pay for, set before execution. It is NOT the// gas actually consumed. Naming this field gasUsed would silently// mislabel every transaction's execution cost as its authorized ceiling. gasLimit:parseInt(tx.gas,16),})),});}// Actual gas consumed only exists post-execution, in the transaction receipt —// a separate RPC call, not a field on the block or its embedded transactions.asyncfunctiongetGasUsed(txHash:string, rpc: EthRpcClient):Promise<number>{const receipt =awaitrpc.call('eth_getTransactionReceipt',[txHash]);returnparseInt(receipt.gasUsed,16);// this is the real, post-execution figure}