TypeScriptでCQRS & Event Sourcing

中村修太

2020.12.11

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

Introduction

一般的になってきたCQRSとEvent Sourcing。
これらのアーキテクチャは、RDBMSやNoSQLデータベースをデータストアとして持っている、
昔ながらのCRUDデータモデルの代案として近年採用が進んでいます。

本稿ではCQRS・EventSourcingの簡単な説明と、
それらを用いたサンプルをTypeScriptで実装してみます。

What is CQRS?

CRUDと違い、CQRS(Command and Query Responsibility Segregation)は
データの読み込みと書き込みは本来違うものである、という前提に基づく考え方で、
2010年にGreg yang氏が考案していたものです。

あらゆるメソッドは、アクションを実行するコマンドか、
呼び出し元にデータを返すクエリかのいずれかであって、両方を行ってはならない。
これは、質問をすることで回答を変化させてはならないということだ。
by Greg yang

CQRSでは、データストアの操作をコマンド（登録・更新・削除）と
クエリ（参照）に分類します。
コマンドは基本、システムの状態を変更して、操作の結果を返します。
クエリは冪等であり、システムの状態を変更せずに結果を返します。
間にコマンドが実行されない限り、同じクエリであれば結果は同じになります。

クエリとコマンド、2つの役割を分離することでシンプルなアプリケーションを構築でき、
例えば書き込み用・読み込み用に
それぞれ異なるデータベースを持つケースなどで使われています。

CQRSにするメリットとして、コマンドとクエリを分離しないとQueryが複雑になったりとか、
オブジェクトデータストア間のインピーダンスミスマッチがあったりとかがありますが、
詳しくはこのへんを参照してください。

ちなみに、弊社で開発してるEC/CRMプラットフォーム、Prismatixにおいても
一部CQRSを取り入れており、RDBMS更新時に参照用データストア(Elasticsearch)へ
データを同期して検索はそちらで対応させる、というようなことを行っています。
※Prismatixについての詳細はこちら

What is Event Sourcing?

Event Sourcingでは、状態(ステート)ではなくイベントを中心としてデータを扱います。
データストアにデータの「現在の状態」だけを格納するのではなく、
そのデータに対して実行された一連のすべての実行を記録しておきます。

RDBMSをデータストアとしたCRUDモデルを例にすると、

登録時にはinsertでDBにレコードを作成
更新時はupdateでDBのレコードを更新
削除時はdeleteでDBからレコードを削除

というような処理が普通だと思います。
１件データ登録後にそのデータの更新が行われれた場合、
最新状態のデータがDBに1件存在します。

これに対してEvent Sourcingだと、
すべての実行を追記形式で記録していきます。
「１件データが登録された」という記録と、
「対象データが○○に更新された」という記録が保存されるわけです。
そのため、対象データの最新状態を取得するには、
登録された記録と更新された記録が必要になります。

Event Sourcingのメリット・デメリットは下記。

メリット

完全な履歴を保存可能
任意の過去の時点に戻れる

デメリット

現在の状態を得るためにログをたどる必要がある
イベントを全て保存するため、ログが大きくなりがち
イベント欠損時のリカバリが難しい
バージョンアップ時の整合性を考慮する

また、Event Sourcing単体で使うすると、
クエリメソッドが複雑になったり、インピーダンスミスマッチが発生したりとかの
問題が発生しやすいのですが、CQRS + Event Sourcingにすることで、
それらの問題も解決できます。

Event Sourcingについての詳細はこのへんとかこのへんを参照してください。

Sample Architecture

CQRSとEvent Sourcingの簡単な説明をしたところで、
表題のサンプルアプリを実装してみます。構成は下記。

User情報の登録・更新・取得用のWeb APIを実装します。表の左(Create、Update)がコマンド、右(Query)がクエリです。
コマンドが実行されるとUser情報がredis-streamsを通じてSqliteに同期されます。

Environment

OS : MacOS 10.15.7
node : v14.4.0
sqlite3 : 3.28.0
redis : 6.0.9

node、redis、sqlite3はHomebrewでインストール済み。

setup

まずは必要なモジュールのインストールです。
TypeScriptに加え、RedisやSqlite用のモジュールをインストールします。
Webサーバとしての機能はnode最速といわれるfastifyを使用します。

% npm install --save fastify
% npm install --save ioredis
% npm install --save sqlite3

% npm install --save-dev @types/node 
% npm install --save-dev @types/ioredis
% npm install --save-dev@types/sqlite3
% npm install --save-dev typescript
# tsc --initしておく

sqlite3でテーブル作成しておきます。

% sqlite3 cqrs.db
sqlite > create table user (code text primary key,name text, mail text,create_timestamp text,update_timestamp text);

redisも起動しておきましょう。

% redis-server

Implement Web API

まずはコマンドやクエリで使用するUser用のクラスや共通インターフェイスを定義します。
※余計なimportとかは省略してます

export interface Model {}
export interface Query {}

//コマンド用Userモデル
export class User implements Model {
  constructor(public code: string, public name: string, public mail: string) {}
}

//クエリ用Userモデル
export class QueryUser implements Query {
  constructor(
    public code: string,
    public name: string,
    public mail: string,
    public create_timestamp: string,
    public update_timestamp: string
  ) {}
}

コマンド実行時にRedisにpublishするクラスを定義します。

import Redis from 'ioredis'

export class RedisUtil {
  private static instance: RedisUtil
  private redis: any

  private constructor() {
    this.redis = new Redis()
  }
  static getInstance() {
    if (!RedisUtil.instance) {
      RedisUtil.instance = new RedisUtil()
    }
    return RedisUtil.instance
  }

  async publish(
    stream: string,
    eventName: string,
    code: string,
    model: Model
  ): Promise<string> {
    const id = await this.redis.xadd(
      stream,
      '*',
      'eventName',
      eventName,
      'code',
      code,
      'json',
      JSON.stringify(model)
    )
    return id.toString()
  }
}

xadd関数を使えば指定したstreamに対してデータを追加します。
イベント名とJson形式のデータをredis-streamsに送ります。

サーバのエントリポイントとなるindex.tsの定義です。
POST(Create)、PUT(Update) GET(Query)用の処理をそれぞれ定義します。

import fastify from 'fastify'
import sqlite3 from 'sqlite3'

interface ICreateUserBody {
  code: string
  name: string
  mail: string
}

interface IUpdateUserBody {
  code: string
  name: string
  mail: string
}

interface IGetUserParam {
  code: string
}

const server = fastify()

server.get<{ Params: IGetUserParam }>('/user/:code', async (request, reply) => {
  const code: string = request.params.code
  const db = new sqlite3.Database('cqrs.db')
  //get data from sqlite3 by code
  db.serialize(function () {
    const stmt = db.prepare(
      'select code,name, mail, create_timestamp,update_timestamp from user where code =?'
    )
    stmt.get(code, function (err, row) {
      if (err) {
        throw err
      }
      const qu: QueryUser = new QueryUser(
        row.code,
        row.name,
        row.mail,
        row.create_timestamp,
        row.update_timestamp
      )
      reply
        .code(200)
        .header('Content-Type', 'application/json; charset=utf-8')
        .send(qu)
    })
  })
})

server.post<{ Body: ICreateUserBody }>('/user', async (request, reply) => {
  const ins: RedisUtil = RedisUtil.getInstance()
  const user: User = request.body
  const publishId: string = await ins.publish(
    'user-stream',
    'UserCreated',
    user.code,
    user
  )
  reply
    .code(200)
    .header('Content-Type', 'application/json; charset=utf-8')
    .send({ eventType: 'UserCreated', id: publishId })
})

server.put<{ Body: IUpdateUserBody }>('/user', async (request, reply) => {
  const ins: RedisUtil = RedisUtil.getInstance()
  const user: User = request.body
  const publishId: string = await ins.publish(
    'user-stream',
    'UserUpdated',
    user.code,
    user
  )
  reply
    .code(200)
    .header('Content-Type', 'application/json; charset=utf-8')
    .send({ eventType: 'UserUpdated', id: publishId })
})

server.listen(8080, (err, address) => {
  if (err) {
    console.error(err)
    process.exit(1)
  }
  console.log(`Server listening at ${address}`)
})

これでWeb APIの準備はOKです。

Implement redis-streams to sqlite program

次はredis-streamsを見てsqliteへ更新するプログラムです。
redis-streamsにデータが追加されるたびにsqliteへinsertかupdateを実行します。

//read-redis-stream.ts
import Redis from 'ioredis'
const readRedis = new Redis()
import sqlite3 from 'sqlite3'

//UserCreatedイベント時にsqliteへinsert
function insert(dbName: string, json: string) {
  const db = new sqlite3.Database(dbName)
  db.serialize(function () {
    const now = "datetime('now', '+9 hours')"
    const sql =
      'insert into user (code, name,mail,create_timestamp,update_timestamp) VALUES (?,?,?,' +
      now +
      ',' +
      now +
      ');'
    const stmt = db.prepare(sql)
    const jsonObj = JSON.parse(json)
    stmt.run(jsonObj.code, jsonObj.name, jsonObj.mail)
    stmt.finalize()
    console.log('insert to sqlite : ' + jsonObj.code)
  })
  db.close()
}

//UserUpdatedイベント時にsqliteへupdate
function update(dbName: string, json: string) {
  const db = new sqlite3.Database(dbName)
  db.serialize(function () {
    const now = "datetime('now', '+9 hours')"
    const sql =
      'update user set name=?,mail=?,update_timestamp=' + now + 'where code=?;'
    const stmt = db.prepare(sql)
    const jsonObj = JSON.parse(json)
    stmt.run(jsonObj.name, jsonObj.mail, jsonObj.code)
    stmt.finalize()
    console.log('update to sqlite : ' + jsonObj.code)
  })
  db.close()
}

//xreadでredis-streamからデータを読み取る
async function subscribeStream(stream: string) {
  let lastID = '$'
  while (true) {
    const reply = await readRedis.xread('BLOCK', 10000, 'STREAMS', stream, '$')
    if (!reply) {
      continue;
    }
    const results = reply[0][1]
    const { length } = results
    if (!results.length) {
      continue;
    }
    lastID = results[length - 1][0]
    //update sqlite
    for (const row of results) {
      console.log('eventName :' + row[1][1])
      if (row[1][1] === 'UserCreated') {
        insert('cqrs.db', row[1][5])
      } else if (row[1][1] === 'UserUpdated') {
        update('cqrs.db', row[1][5])
      }
    }
  }
}

subscribeStream('user-stream');

このプログラムは、Webサーバとは別に起動しておく必要があります。

Execute Program

プログラムの記述ができたので動作確認をします。
まずはpackage.jsonに↓のscriptsを追記します。

・・・ 
 "scripts": {
    "build": "tsc -p tsconfig.json",
    "start": "node index.js"
 },
・・・

buildコマンドでTypescriptのコンパイル。

% npm run build

startコマンドでWebサーバを起動します。

% npm run start

redis-streamsからデータを取得してsqliteへ更新するプログラムの起動もしておきます。

% node read-redis-streams.js

curlを使ってAPIにアクセスしてみます。
POST APIでユーザーの登録を実行。

% curl -X POST -H "Content-Type: application/json" -d '{"code" : "USER0001","name":"syuta", "mail":"syuta@cm.com"}' localhost:8080/user
{"eventType":"UserCreated","id":"1607687729494-0"}

PUT　APIで登録したユーザーの更新。

% curl -X PUT -H "Content-Type: application/json" -d '{"code" : "USER0001","name":"syuta-update", "mail":"syuta@cm.com"}' localhost:8080/user
{"eventType":"UserUpdated","id":"1607687906732-0"}%

クエリを実行すると最新状態のユーザー情報が取得できます。

% curl -X GET localhost:8080/user/USER0001
{"code":"USER0001","name":"syuta-update","mail":"syuta@cm.com","create_timestamp":"2020-12-11 20:55:29","update_timestamp":"2020-12-11 21:04:46"}

Summary

今回はCQRS ＋ Event Sourcingの簡単なサンプルを実装してみました。
実際にCQRS+ESで実装できるフレームワークとして、
AxonやAkkatecture、Lagomなどがあるので、
興味があるかたはそれらも確認してみてください。

TypeScriptでCQRS & Event Sourcing

Introduction

What is CQRS?

What is Event Sourcing?

Sample Architecture

Environment

setup

Implement Web API

Implement redis-streams to sqlite program

Execute Program

Summary

Reference

イベント

EVENT【5/28（火）】AWSを最大活用するための1dayカンファレンス

EVENT【5/17（金）】認証機能の開発工数削減をデモで体験！次世代認証基盤サービス『Auth0 by Okta』導入実践ウェビナー

EVENT【5/14（火）】アノテーションの中途向けオンライン会社説明会を開催します

EVENT【5/23（木）】ユースケースに学ぶAWS運用のノウハウ～可視化から統制まで～

EVENT【5/16（木）】Snowflakeを触ってみよう！初めての方向けハンズオンセミナー

EVENT【4/18リモート】クラスメソッドのフリーランスエンジニア会社説明会〜AWS/システムアーキテクト/Android案件特集〜を開催します

EVENT【5/10（金）名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【4/25（木）リモート】Google Cloudに携わるエンジニアのキャリア～クラスメソッド会社説明会～

EVENT【4/30（火）リモート】クラスメソッドの会社説明会を開催します

EVENT【5/9（木）】組織的なクラウド統制はじめの一歩

TypeScriptでCQRS & Event Sourcing

Introduction

What is CQRS?

What is Event Sourcing?

Sample Architecture

Environment

setup

Implement Web API

Implement redis-streams to sqlite program

Execute Program

Summary

Reference

イベント

EVENT【5/28（火）】AWSを最大活用するための1dayカンファレンス

EVENT【5/17（金）】認証機能の開発工数削減をデモで体験！次世代認証基盤サービス『Auth0 by Okta』導入実践ウェビナー

EVENT【5/14（火）】アノテーションの中途向けオンライン会社説明会を開催します

EVENT【5/23（木）】ユースケースに学ぶAWS運用のノウハウ～可視化から統制まで～

EVENT【5/16（木）】Snowflakeを触ってみよう！初めての方向けハンズオンセミナー

EVENT【4/18リモート】クラスメソッドのフリーランスエンジニア会社説明会 〜AWS/システムアーキテクト/Android案件特集〜 を開催します

EVENT【5/10（金） 名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【4/25（木）リモート】Google Cloudに携わるエンジニアのキャリア～クラスメソッド会社説明会～

EVENT【4/30（火）リモート】クラスメソッドの会社説明会を開催します

EVENT【5/9（木）】組織的なクラウド統制 はじめの一歩

関連記事

ElectricSQLでLocal-firstなアプリを作る

Vite + Vitest の環境を設定してみた

Cloudflare Workers間でRPC通信する

type-challengesでTypeScriptの型を学んでみた【初級編】

EVENT【4/18リモート】クラスメソッドのフリーランスエンジニア会社説明会〜AWS/システムアーキテクト/Android案件特集〜を開催します

EVENT【5/10（金）名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【5/9（木）】組織的なクラウド統制はじめの一歩