小規模アプリでのCQRS+イベントソーシングの構成例

# 経緯
CQRS+イベントソーシングをオンプレミスで実装しようとしたときに、
Apache Kafka等のメッセージキューとかを使わずにイベント検知できないかと考えたところ、
PostgreSQLのPub/Sub機能を使えば最小限で出来そうな気がしたので検討アプリを作ってみました。
アプリは[ここ](https://github.com/kwhrkzk/postgres_eventsourcing_sample)に置いています。

cloneしたら

```sh
cd src
docker-compose up -d --build
```

で起動します。

# 環境
* [自作検討アプリ](https://github.com/kwhrkzk/postgres_eventsourcing_sample)
* docker-compose
* mcr.microsoft.com/dotnet/sdk:7.0
* Microsoft.EntityFrameworkCore Version=7.0.0-preview.7.22376.2
* postgres:14.5-bullseye

# 環境イメージ

* write_model(eventstore用のpostgres)
* read_model(参照用のpostgres)
* Projector(read_modelへイベントを射影するdotnetcore)
* SampleCmd(write_modelへイベントを追加するdotnetcore)

![名称未設定ファイル.drawio.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109302/96b72196-becd-1b74-198e-059820110b55.png)

## ファイルツリー抜粋

```sh
├── docker-compose.yml
├── dotnet
│   ├── Dockerfile
│   ├── docker-entrypoint.sh
│   └── src
│       ├── Domain
│       │   ├── Domain.csproj
│       │   ├── EventIdEnum.cs
│       │   ├── IEventInsert.cs
│       │   ├── User.cs
│       │   ├── UserId.cs
│       │   ├── UserName.cs
│       ├── Infrastructure
│       │   ├── DataBase
│       │   │   ├── Event.cs
│       │   │   └── EventstoreContext.cs
│       │   ├── EventInsert.cs
│       │   ├── Infrastructure.csproj
│       ├── Projector
│       │   ├── Commands.cs
│       │   ├── DataBase
│       │   │   ├── SampleDBContext.cs
│       │   │   └── User.cs
│       │   ├── Program.cs
│       │   ├── Projector.csproj
│       └── SampleCmd
│           ├── Commands.cs
│           ├── DataBase
│           │   ├── SampleContext.cs
│           │   └── User.cs
│           ├── Program.cs
│           ├── SampleCmd.csproj
├── read_model
│   └── Dockerfile
└── write_model
    ├── Dockerfile
    └── initdb
        ├── eventstore.backup
        └── init.sh
```

# ドメインとかイベントとか

とりあえずただのユーザーテーブルにCRUDするドメインにしてみました。

## ドメイン
```plantuml
@startuml
class UserName {
string name
}
class UserId {
Guid id
}
class User {
UserId ID
UserName Name
}
User -- UserId
User -- UserName
enum EventIdEnum {
UserAdded = 1
UserEdited = 2
UserDeleted = 3
}
@enduml
```

## イベント
| EventId | 名称 | 説明 | スキーマっぽいもの |
|:-:|:-:|:-:|:-:|
| 1 | UserAdded | ユーザーを追加する | { "type": "object", "properties": { "name": { "type": "string" } } } |
| 2 | UserEdited | ユーザーを編集する | { "type": "object", "properties": { "id": { "type": "string" }, "name": { "type": "string" } } } |
| 3 | UserDeleted | ユーザーを削除する | { "type": "object", "properties": { "id": { "type": "string" } } } |

## ER図
```plantuml
@startuml
entity Events {
* id uuid
* event_id int
* data jsonb
* meta jsonb
* created timestamp
}
note top: write_modelの\neventstoreデータベースに\n用意
entity Users {
* id uuid
* name text
* created timestamp
modified timestamp
}
note top: read_modelの\nsampleデータベースに\nProjectorが作成
@enduml
```

# Projectorのコマンド
データベースリストア用のinitと、イベント購読用のsubscriptionを用意しました。

```sh
$ docker exec -it dotnet dotnet run --project /root/src/Projector help
Usage: Projector <Command>

Commands:
  help            Display help.
  init            
  subscription    
  version         Display version.
```

## データベースリストア用のinitの中身抜粋

```csharp
    // read_modelのSampleデータベースのDbContextをコンストラクタインジェクションしている.
    private SampleDBContext Smp { get; }

    // write_modelのEventstoreデータベースのDbContextをコンストラクタインジェクションしている.
    private EventstoreContext Ev { get; }

    public async Task Init()
    {
        Smp.Database.EnsureDeleted(); // Sampleデータベースの削除
        Smp.Database.EnsureCreated(); // Sampleデータベースの生成

        // イベント分再生
        foreach (var x in Ev.Events)
            Handle(x);
    }

    // イベントid毎に処理.usersテーブルにinsertしたりupdateしたりdeleteしたり.
    private void Handle(Event ev)
    {
        if (Enum.IsDefined(typeof(EventIdEnum), ev.EventId) == false)
            return;

        switch ((EventIdEnum)ev.EventId)
        {
            case EventIdEnum.UserAdded: UserUpsertHandler(ev.Data.RootElement); break;
            case EventIdEnum.UserEdited: UserUpsertHandler(ev.Data.RootElement); break;
            case EventIdEnum.UserDeleted: UserDeleteHandler(ev.Data.RootElement); break;
            default: throw new ArgumentException($"not implemented. {ev.EventId}");
        };
    }
```

## イベント購読用のsubscriptionの中身抜粋

```csharp
    public async Task Subscription(EventstoreContext ev)
    {
        var conn = (NpgsqlConnection)ev.Database.GetDbConnection();
        conn.Open();

        try
        {
            conn.Notification += OnNotification; 

            // postgresqlのLISTENで購読する.
            using var cmd = new NpgsqlCommand("LISTEN event_channel", conn);
            cmd.ExecuteNonQuery();

            while (Context.CancellationToken.IsCancellationRequested == false)
                await conn.WaitAsync(Context.CancellationToken);
        }
        catch (Exception ex) when (!(ex is OperationCanceledException))
        {
            Console.WriteLine(ex.ToString());
        }
        finally
        {
            conn.Close();
        }
    }

    // イベントが発行される度に処理.
    private void OnNotification(object sender, NpgsqlNotificationEventArgs e)
    {
        Console.WriteLine("event handled: " + e.Payload);
        Event? ev = JsonSerializer.Deserialize<Event>(e.Payload, new JsonSerializerOptions 
        {
            PropertyNameCaseInsensitive = true,
        });

        if (ev is not null)
            Handle(ev);
    }
```

# Eventstoreデータベースのトリガ部分
write_modelのデータベースに仕込みます。

トリガー条件: eventsテーブルにINSERT
実行内容: insertされた内容をNOTIFY


```sql
CREATE OR REPLACE FUNCTION public.event_insert_trigger()
    RETURNS trigger
    LANGUAGE 'plpgsql'
    COST 100
    VOLATILE NOT LEAKPROOF
AS $BODY$
BEGIN
PERFORM pg_notify(
	'event_channel',
	row_to_json(NEW)::text
);

return NULL;
END
$BODY$;

ALTER FUNCTION public.event_insert_trigger()
    OWNER TO postgres;

CREATE TRIGGER event_trigger
    AFTER INSERT
    ON public.events
    FOR EACH ROW
    EXECUTE FUNCTION public.event_insert_trigger();
```

# SampleCmdのコマンド
とりあえずコンソールアプリで。addとかeditとかイベント名を指定して必要な引数を付けてイベントをwrite_modelにinsertします。

```sh
$ docker exec -it dotnet dotnet run --project /root/src/SampleCmd help
Usage: SampleCmd [options...]

Options:
  -e, --eventname <String>    event name. add or edit or delete (Required)
  -n, --username <String>     user name. Required for add or edit (Default: null)
  -id, --id <String>          guid. Required for edit or delete (Default: null)

Commands:
  help       Display help.
  version    Display version.
```

# まとめ
とりあえず動作しました。
read_modelを増やす予定がないのなら
![index.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109302/a50f1bf2-0fc1-7562-3949-b184aa382091.png)

のようにひとつのPostgresqlでスキーマで分けてしまえば良い気もします。

Projectorの死活監視が面倒なら
![index.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109302/3460e91b-44bb-e3a3-5544-ea5d0b312a03.png)

のようにイベント発行のタイミングでProjectorを外部アプリとして実行するのも有りな気がします。
このパターンでもread_modelをリストアする際にwrite_modelの情報は必要です。

オンプレミスでCQRS+イベントソーシングで実装する際に便利な仕組み等あればご教示願います。