Whoz
Let’s build an Event Sourcing app with DDD pt.1
Tung V
  2024-09-29 00:00:00   2024-10-14 17:39:18  8 Mins

Recently, there has been a trend in Event Sourcing and Domain-Driven Design patterns like the hype of microservices 10 years ago.

And I’m lucky enough to build a rich feature application using event sourcing with in-memory state (Redux) in my previous project, so I can understand some of the pros and cons. But to me the cons is overweight the pro in that case.

So, I decided to make a simple billing app that better suits the ES goal and allows you to get a glimpse of what it is like.

Board the hype train!

*Disclaimer: this is not a production-ready or proper architecture design; it’s a brand new thing I’m experimenting with and not the same as the one I worked on in a previous project.

_You can pull the code here and run on local
_https://github.com/tungvotan/sample-event-sourcing-app

The app

  • A billing app because it’s a simple operation, not many different states and complex business logic, which is ideal for ES or blockchain-like applications.

  • We will go over it step by step, use the hybrid approach first to simulate the product development life cycle. We will make it more event sourced as we go. The reason why is implementing ES from the start will cause a lot of issues and we want to priority the stability and feature delivery.

  • Allows users to open accounts, add funds and charge. ← This will be covered in this part.

  • Replaying events feature to regenerate data from the event stream.

  • Add invoice feature and other things with Kafka.

  • Snapshotting.

  • Eventual consistency lol

Tech stack

  • Node.js, TypeScript, PostgreSQL.

  • Not using an in-memory state (pure JSON object, e.g reducer store) because of a lot of disadvantages like

  • Unable to use multiple instances because it’s impossible to keep the in-memory to be in sync.
  • Max heap size.
  • Perf issue, and we have to reinvent the wheel for implementing the index, proxy (view)… any RDBMS already supports that.

Directory Structure

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
billing-app/
├── db/
│ ├── dbClient.ts
│ └── withTransaction.ts
├── domain/
│ ├── commands/
│ │ ├── addFunds.ts
│ │ ├── openAccount.ts
│ │ └── charge.ts
│ └── events/
│ │ └── accountEvent.ts
│ └── entities/
│   └── accountEntity.ts
│ └── operations/
│   └── account.ts
├── application/
│ ├── accountService.ts
├── repo/
│ ├── accountRepo.ts
│ ├── eventRepo.ts
├── index.ts
└── package.json

The Database

It will be just PostgresDB with two tables

  • accounts: for storing the account information.

  • account_events: for storing events for the account entity.

The Event

We are going to only have 3 simple events for now:

  • ACCOUNT_CREATED: to open a new account

  • FUND_ADDED: add funds to the account (increase the balance)

  • CHARGE_APPLIED: charge the account (decrease the balance)

The event name follows the format NOUNCE_VERB to indicate the action taken.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
export type eventType = 'ACCOUNT_CREATED' | 'FUND_ADDED' | 'CHARGE_APPLIED';

export type AccountEvent = {
  accountId: string;
  eventType: eventType;
  payload: any;
  version: number;
  createdAt?: Date;
};

export const createAccountEvent = (
  accountId: string,
  eventType: eventType,
  payload: any,
  version: number
): AccountEvent => ({
  accountId,
  eventType,
  payload,
  version,
});

The Domain

For simplicity’s sake, let’s start with one simple entity: Account.

The entity schema will be like:

1
2
3
4
5
export type Account = {
  accountId: string;
  balance: number;
  version: number; // this is for the future snapshot feature
};

And we add the basic account operation, that can be used in the command and when replaying events.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
export const createAccount = (accountId: string, balance = 0): Account => ({
  accountId,
  balance,
  version: 0,
});

export const addFunds = (account: Account, amount: number): Account => ({
  ...account,
  balance: account.balance + amount,
  version: account.version + 1,
});

export const charge = (account: Account, amount: number): Account => {
  if (amount > account.balance) throw new Error('Insufficient funds');
  return {
    ...account,
    balance: account.balance - amount,
    version: account.version + 1,
  };
};

The Repository

The repository serves as a bridge between the domain and the data persistence layers. It provides a clean abstraction for accessing and managing. This makes the code easier to test, maintain, and scale because the domain logic can focus on business rules while the repository handles the technical side of data storage. It’s like a helpful middleman that keeps everything tidy and well-separated.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import db from '../db/dbClient';
import { Account } from '../domain/entities/accountEntity';
import { AccountEvent } from '../domain/events/accountEvent';

export class AccountRepository {
  async getById(accountId: string): Promise<Account | undefined> {
    const result = await db
      .selectFrom('accounts')
      .selectAll()
      .where('account_id', '=', accountId)
      .executeTakeFirst();
    //  TODO: put a transform snake-case to camelCase
    return (
      result && {
        ...result,
        accountId: result.account_id,
        version: result.version,
      }
    );
  }

  async save(account: Account): Promise<void> {
    await db
      .insertInto('accounts')
      .values({
        account_id: account.accountId,
        balance: account.balance,
        version: account.version,
      })
      .onConflict((oc) =>
        oc.column('account_id').doUpdateSet({
          balance: account.balance,
          version: account.version,
        })
      )
      .execute();
  }
}

The commands

Then, we will have a simple command handler like handleAddFunds , do the same for handleOpenAccount and handleCharge .

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
export const handleAddFunds = async (
  command: AddFundCommand,
  accountRepository: AccountRepository,
  eventRepository: EventRepository
) => {
  const { accountId, amount } = command;

  const account = await accountRepository.getById(accountId);
  if (!account) {
    throw new Error(`Account with ID ${accountId} does not exist.`);
  }
  const event = createAccountEvent(
    accountId,
    'FUND_ADDED',
    { amount },
    updatedAccount.version
  );
  await eventRepository.saveEvent(event);

  // There is something wrong about this, I will cover in the next part
  const updatedAccount = addFunds(account, amount);
  await accountRepository.save(updatedAccount);
};

So, in this handler, we will retrieve the account data and then use the addFunds operation to generate the new account data.

After that, we save both the event and account data in the accounts and account_events tables, and they must be wrapped in transactions to make sure the account entity and account events are in sync for now since we don’t utilize the ES that much. Let’s make sure the data is correct first then we can toggle it.

*Something is really wrong with this implementation that I’ll reveal in the next part. Have you found the nonsense part I put in here?

The service

I jumped the gun when creating this, it will be covered in the next part because we will have to utilize the Kafka stream to route the command to its handler.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
export const processCommand = async (
  command: any,
  accountRepository: AccountRepository,
  eventRepository: EventRepository
) => {
  switch (command.commandType) {
    case 'OPEN_ACCOUNT':
      await handleOpenAccount(
        command as OpenAccountCommand,
        accountRepository,
        eventRepository
      );
      break;
    case 'ADD_FUND':
      await handleAddFund(
        command as AddFundCommand,
        accountRepository,
        eventRepository
      );
      break;
    case 'CHARGE':
      await handleCharge(
        command as ChargeCommand,
        accountRepository,
        eventRepository
      );
      break;
    default:
      throw new Error(`Unknown command type: ${command.commandType}`);
  }
};

However in this example, we can skip it and call the command handler directly in the index.ts

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
const openAccount = async (accountId: string) => {
  await withTransaction(async function openAccount(transaction) {
    // replace this with handleOpenAccount fn
    await processCommand(
      { commandType: 'OPEN_ACCOUNT', accountId },
      accountRepository,
      eventRepository
    );
  });
};

app.post('/api/account', async (req, res) => {
  const { accountId } = req.body;

  try {
    await openAccount(accountId);
    res.status(200).send({ message: `Opened account ${accountId}` });
  } catch (error: any) {
    console.error(error);
    res.status(500).send({ error: error.message });
  }
});

Let’s test it!

I made some requests on my local and got this result in 2 tables

This is the event stream, which we can use to rebuild the account data when needed.

This is the account data

It’s working!

Conclusion

In this first part, we’ve laid the groundwork for building an event-sourcing app using Domain-Driven Design. By structuring our system around events and aggregates, we’ve taken the first step toward building a scalable, maintainable application that mirrors real-world business logic. There’s a lot more to cover — like handling projections, snapshots, and eventual consistency — but this foundation sets the stage for deeper exploration. Keep an eye out for the next part as we dive deeper into more advanced patterns and practical implementations!