humaとsqlcでCQRS・ESに入門してみた

はじめに

こんにちは。富士通の阿部です。こちらの記事は Fujitsu Advent Calendar 2024 の 23日目 の記事です。

本アドベントカレンダー内では、量子計算やFortranの記事など興味のある記事がたくさん出ていたので、後でじっくり読んでみようと思います。

わたしの記事では敷居を下げて「入門してみた」の話になります。よろしくお願いします。

概要

下記の要素に触れつつ簡単なTODOアプリを作成しました。

• sqlc
• CQRS・ES
• Huma
• Goの状態遷移管理をenumでやる

きっかけ

こちらの本を読んで

• DDD + CQRS・ESをやってみたくなった
• sqlcはGoでDB触る機会があればやってみたかった
• Humaはこの記事を書く前にちょっとバズっていて気になったので急遽

アプリ概要

簡単なタスク管理ツールになります

• データベース: postgreSQLをsqlcで利用
• Webページ: ビルトインのhttpパッケージのhttp.ServeMuxを利用
• API: http.ServeMuxをHumaでwrapして利用
• アーキテクチャ: CQRS・ESを利用してイベントストアにイベントを収集 + イベントをもとにTaskProjectionで現在のタスクの状態を反映
• UI: とりあえずで

いったんは勉強用で作ったものですが、あわよくば普段のタスク管理+TODO管理ツールとして使えるように更新していきたいです。

各要素の使い勝手

sqlc

sqlcではORMを覚える必要がなく、直接SQLを操作する感覚がシンプル感あって好みでした。

クエリ定義ファイル

-- name: UpdateTaskState :exec
UPDATE tasks SET status = $2
WHERE task_id = $1;

sqlcの利用コード

// postgresのconn作成
ctx := context.Background()
connString := "postgres://root:password@localhost:5432/appdb?sslmode=disable"
conn, err := pgx.Connect(ctx, connString)
if err != nil {
	slog.Error("postgres connection error", slog.Any("err", err))
}
defer conn.Close(ctx)

// sqlcの利用
queries := db.New(conn)

queries.UpdateTaskState(ctx, db.UpdateTaskStateParams{
  TaskID: uuid.New(),
  Status: db.TaskStatus("hogehoge"),
);

実際のコーディングの流れとしては、定義ファイルに-- name: UpdateTaskState :execのように記述して、go generate ./...した後、コード上で、queries.まで打って補完でどんどん埋めていくだけなので手軽でした。

今回はまだ簡単なクエリを少量しか書けていないので、複雑なクエリを書いてみたいです。クエリをブクブク増やしてゆく運用と、汎用的なクエリにとどめてアプリ側で頑張る運用のどちらが良い塩梅かなど見ていきたいです。

Huma

Humaを利用することで、Goのhttp.ServeMuxを拡張しながら、APIエンドポイントを定義し、OpenAPI 3.0/3.1仕様のドキュメントを自動生成できました。コードだとこの辺
OpenAPI 3.0/3.1仕様のドキュメントを吐き出せるのは2024/12月現在humaだけらしいです。

ビルトインのハンドラ関数は、

type RequestHoge struct {...}
type ResponseHoge struct {...}
func HogeHandler(w http.ResponseWriter, r *http.Request) {
    var req RequestHoge
    json.NewDecoder(r.Body).Decode(&req)
    ...
    resp := ...
    json.NewEncoder(w).Encode(resp)
}

のような形式で、I/Oストリームを意識したデータハンドリングが必要です。（Goでストリームを上手に使っているコードはカッコよくて結構好きです。）
一方、Humaでは

type RequestHoge struct {...}
type ResponseHoge struct {...}
func HogeHandler(ctx context.Context, req *RequestHoge) (*ResponseHoge, error) {
...
}

という形式で、普通の関数に似ていて親しみやすいです。

ハンドラ関数をrouterに登録部分は、ビルトインのhttp.ServeMuxの場合は、

mux.HandlerFunc("POST /tasks/{taskId}", HogeHandler)

の形ですが、Humaでは、

api := humago.New(mux, huma.DefaultConfig("TaskTODO", "1.0.0")
huma.POST(api, "/tasks/{taskId}", HogeHandler)

という形であまり変わらないです。Go 1.22でビルトインのmuxも使い勝手がかなり良くなっているので、正直ビルトインのほうが好みです。（http.ServeMuxで利用できるpathのパターン）

それよりもHumaを使いたかった理由として、open API 3.0/3.1準拠のドキュメントページが/docに自動生成されるというのがあります。

ドキュメントの内容は、以下のようにRequest/Responseに利用する構造体のタグに記載します。

type Task struct {
	Name   string    `json:"name" example:"空き缶を片付ける" doc:"task name"`
	Id     uuid.UUID `json:"task_id" example:"ea087856-822f-4db3-a9c7-23ce3dd337b3" doc:"uuid"`
	Status string    `json:"status" enum:"pending,doing,completed,cancelled" example:"pending" doc:"task status"`
}

type ResponseTasks struct {
	Body []Task
}

（ここのコードはリクエストボディしかRequestの構造に含まれていないが、クエリパラメータ、パスパラメータなどもRequest内で定義していく）
ここのタグを丁寧に書いていくことで、OpenAPIのドキュメントが充実していきつつ、内容によっては、enumとかformat: "uuid"とかを書いてあげることでハンドラー関数の処理の手前でバリデーションが実施されます。

あまり手をかけなくてもサマになるので、書いていて楽しかったです。もっと内容を充実させたいなーと思いながら書いていました。
Humaの利用もチュートリアルのレベルなので、HumaでできることはなるべくHumaにやらせる方針で使っていきたいです。

DDD

Goでenum+状態遷移の表現

goでenum（っぽいもの）を作る方法はいくつかあり、素朴にstringで列挙したり、type MyType stringして列挙したり、以下のようにコード生成を用いたり

stringer

enumer

enumerはstringerに便利機能大盛りみたいなやつです。今回はenumerを使っています。

状態をenumで表現して、状態1から状態2に遷移可能か？を判定したいときがあります。そんな時はこんな風にenumの二次元setの形で書くと便利です。

// 状態遷移を表現するジェネリック型で、comparable型なら何でもイケる
// 外側のmapのキーが「遷移元(from)」の状態、内側のmapのキーが「遷移先(to)」の状態を表す
// 内側のmapの値は、遷移が許可される場合にtrueとなる
// 値のboolはstruct{}でもよい。お好みで
type transitionType[T comparable] map[T]map[T]bool

// 指定されたfromからtoへの遷移が有効かどうかを判定する関数
// 遷移可能ならtrue, 不可能ならfalseを返す
func (t transitionType[T]) CanTransition(from, to T) bool {
	if toMap, exists := t[from]; exists {
		if canTransition, ok := toMap[to]; ok {
			return canTransition
		}
	}
	// 遷移が定義されていない場合や許可されていない場合はfalseを返す
	return false
}

// 生成関数
// fromをキー、toのスライスを値とするマップからtransitionTypeを生成する
// 遷移状態を定義する時に、trueやらstruct{}やら不要なので
func newTransitions[T comparable](entries map[T][]T) transitionType[T] {
	// transitionType構造を初期化
	transitions := make(transitionType[T])
	// 「遷移元(from)」と対応する「遷移先(to)」を反復処理
	for from, tos := range entries {
		transitions[from] = make(map[T]bool)
		// 内側のマップに「遷移先(to)」をtrueとして設定
		for _, to := range tos {
			transitions[from][to] = true
		}
	}
	return transitions
}

// タスクの状態を表す型
// こいつの遷移を表現してみる
type TaskStatus int

// TaskStatusの種類
const (
	Pending TaskStatus = iota   // タスクが保留中で、まだ開始されていない状態
	Doing                       // タスクが現在進行中の状態
	Completed                   // タスクが完了した状態
	Cancelled                   // タスクがキャンセルされた状態
	Unknown                     // タスクの状態が不明または未定義の状態
)

// TaskStatusの遷移の表現
var TaskStatusTransition = newTransitions(map[TaskStatus][]TaskStatus{
	Pending:   {Doing, Completed, Cancelled}, // 保留中(Pending)のタスクは、進行中(Doing)、完了(Completed)、キャンセル(Cancelled)に遷移可能
	Doing:     {Pending, Completed, Cancelled}, // 進行中(Doing)のタスクは、保留中(Pending)、完了(Completed)、キャンセル(Cancelled)に遷移可能
	Completed: {Doing},                      // 完了(Completed)のタスクは、進行中(Doing)にのみ遷移可能
	Cancelled: {},                           // キャンセル(Cancelled)されたタスクは、他の状態に遷移不可能
})
// Unknownのことは無視している、ごめんね。

利用する側は以下のようなコードになる。

type Task struct {
	Name string
	Status TaskStatus
}

func (t *Task) CanTransition(to TaskStatus) bool {
	return TaskStatusTransition.CanTransition(t.Status, to)
}

func (t *Task) Transit(s TaskStatus) (*Task, error) {
	if !t.CanTransition(s) {
		return nil, fmt.Errorf("invalid transition from %v to %v", t.Status, s)
	}
	return &Task{Name: t.Name, Status: s}, nil
}

func StartTaskUseCase(task_id uuid.UUID, newStatus TaskStatus, repo TaskRepository) error {
	task, err := repo.GetTask(task_id)
	if err != nil {...}
	newTask, err := task.Transit(newStatus)
	if err != nil {
		return fmt.Error("task transition error: %w", err)
	}
	err := repo.UpdateTask(task_id, newTask)
	if err != nil {...}
	return nil
}

switch版

func (t *Task) Transit(to TaskStatus) (*Task, error) {
	switch t.Status {
	case Pending:
		if to == Doing || to == Completed || to == Cancelled {
			return &Task{Name: t.Name, Status: to}, nil
		}
	case Doing:
		if to == Pending || to == Completed || to == Cancelled {
			return &Task{Name: t.Name, Status: to}, nil
		}
	case Completed:
		if to == Doing {
			return &Task{Name: t.Name, Status: to}, nil
		}
	case Cancelled:
		return nil, fmt.Errorf("cannot transition from %v to %v: task is cancelled", t.Status, to)
	default:
		return nil, fmt.Errorf("unknown task status: %v", t.Status)
	}

	return nil, fmt.Errorf("invalid transition from %v to %v", t.Status, to)
}

というような感じで、transitionTypeを用いることでスッキリ書けて気持ちが良いですね。

DDDは現在取り組み中のため、どんどん内容追記予定です。

CQRS・ES

CQRSは今までメリットがわからず勉強できていませんでしたが、ES（Event Sourcing）のことを知ったことと、丁度自分のタスク管理方法に悩んでいて、自分のタスクやTODOのこなし具合を統計とってグラフで見たりできたら楽しいだろうな、と思っていたのでそのあたりのピースがハマって入門しよう、ということになりました。

今回のアプリでは、イベントストアテーブル + タスクのプロジェクション用テーブル + チェックポイントのテーブル、という構成でやっています。

実装するだけした状態で、まだあまり頭の中が整理できていないので、整理できたら構成などの詳細を追記していこうと思っています。

今のところの感想としては、実装が重いけど拡張性があるという雰囲気です。
特に、Get側の利用用途に応じたテーブルと投影エンジンを作ることで、共通の情報源から複数のデータを表現できるというのは、監査用・統計用・リソース監視用などの機能を独立性高く管理でき、イベントをPOSTしていく側（コマンド側）の処理もそれらの要件に左右されず安定させられそうで、便利に感じます。

今後の展望

• DDD（ドメイン駆動設計の本を読んだのに、まだドメイン駆動設計していないので）
• テスト（今のところ0個なので）
• 今回の記事では触れませんでしたが、slogとgolang-migrateも入門中なので、いろいろ機能を使ってみたいです。

最後に

最後まで見ていただきありがとうございました。気になるところがあればご連絡ください！