このエントリはスターフェスティバル株式会社のスターフェスティバル Advent Calendar 2022
16日目記事でもあります。

みなさんは開発する時にどう考えていますか？

大した内容ではありませんが、今回は開発をする上で
「どう考えて設計して表現していくか」、という永遠の悩みの中で
自分が複雑な物事に立ち向かう時の頭の中を少し書き出してみようと思います。
各カンファレンスなどで話しているものを結合したものではあります。

一緒に仕事をしたりしている方々にはお馴染みの話です

前半くらいは前提の話や分析の思考、
後半はイベント駆動などにおけるメッセージについて
という流れになってます。

ちなみに自身はスターフェスティバルではアプリケーション全般の開発には関わっていますが、
主にデータ基盤やデータドリブンなマインドを伝播させていくことや、
データを使った戦略を立てながらのプロダクト作りや、インフラ全般に携わっています。
（オフライン勉強会とかも少なくなったのでこういう話をする機会も減りましたねぇ・・）

細かいドメインモデルやデータモデル、設計手法の云々等については触れません（十分なボリュームすぎて・・）。

商品とは？

せっかくなのでスターフェスティバルで取り組んでいるテーマを例にしてみましょう。
*そのものずばりの内容ではなく、実際のものよりも単純にしていたり想像しやすい内容にしています

前提としてスターフェスティバルは下記の領域を取り組んでいる会社です。

企業理念
ごちそうで 人々を より 幸せに

飲食店が中食・デリバリーに参入するためのソリューションとして、
「製造」以外の部分にあたる「商品開発」「販路提供」「販売促進」「注文受付」「決済」
「配達」のすべてをスターフェスティバルが一気通貫でトータルサポートいたします。

また、最大級のモールを運営する弊社ならではの、製造・物流ネットワークや
販売チャネルを活用したサービスも行っています。

何らかのシステムを介して、おいしいごちそうを頼みたい！ユーザー（喫食者と呼んでいます）、
おいしいごちそうを提供したい製造元（以下製造パートナー）と
そしてそれを届ける配送が全国各地でリアルに動く、と大まかな関係があります。

では商品とは何を指すのでしょうか？

ドメイン知識などがない場合は、商品という言葉から
「8割くらいごちクルで販売している弁当とかのことを商品というんだな、
データベースにはこの商品があるだろう」
と思うはずです。

これは購入する喫食者の視点からみた時の認識であると同時に、
開発者としての視点での認識でもあります。

現状を理解せずに喫食者（らしい者）と開発者視点のままで開発をしていくと、
半分は飲食店のメニューとして並んでいるような商品で、
半分は商品の属性などが入り混じった商品という言葉から連想されるものが結合した何か、
データベース設計でいうならばポリモーフィック前提のような構造になってしまいます。

これを読んでいるみなさんの周りのアプリケーションにも似たような状態のものはたくさんあると思います。
アプリケーションがきちんと機能していればネガティブなものではありません。
多少見通しが悪いなどはありますが

これまではこれでも良かった。

が、物事は永遠に変わらないわけではなく、
色々と変化が必要なタイミングもあり業務フローとシステムを大きく見直す必要が出てきました。

取り組んでいる領域には「ごちそう」を軸に

喫食者
製造パートナー
配送者

の3つがあることはわかっていましたが、本当にそれだけでしょうか？

実際は製造以外の分野を引き受けるため、
ごちそうを商品として登録する運用チーム、営業チームがあり、
このチームが製造パートナーの商品情報を受け取り、
サービスが提供する商品として成立するための情報を付与し、
はじめて喫食者が目にする「ごちそう」となります。

よく考えたら当たり前の構造ではありますが、
これまではよくても（雑にいうと）事業成長を支えていくためには現在の会社の状況を正しく理解し、
業務フローもスケールしやすくするためにはある程度ながく戦えるシステム・仕組み（例えです）を作らなければなりません。

事業を支えていくような基盤の場合は短い期間で大きなリプレイスが難しい場合もありますし、
サービスの特性や領域によっては簡単にリプレイスを繰り返すものもあります。
皆さんの環境や領域ではまた変わってきますので、皆さんの環境に置き換えて考えてみてください！
（これはあくまで例です）

業務フローとシステム（みなさんが想像しやすいアプリケーションと思っていただいて良いです）は
密接な関係があり、「運用でカバー」がそのまま謎の進化を遂げたり、
とりあえず今の要求通りで動けばいいやというシステムは、
やがてスケールできない業務フローや事業戦略が立てづらい企業と繋がっていきます。
（そのために継続的なリファクタリングや、変化に対応しやすく未来を一緒に作っていけるアジャイルなどを取り入れたり）

現状をより理解する

では長く戦うために考えることは、
マイクロサービスアーキテクチャ
でしょうか？

いやいや、そうではなくて現在の業務フローや構造などを再度見つめなおしてみます。

「ごちそう」と表現される商品の大元は、
弁当などを製造する店舗・製造パートナーがありそこで認識される商品は、
あくまで製造側が認識できる「xx弁当」という商品になります。

たとえば製造側は上記のような認識だとしましょう。

製造パートナーが認識できる商品に対してスターフェスティバルで必要な情報が付与されます。
配送に関連したものであったりさまざまですが、
店舗が認識している商品とはまた少し違った形となります。

同様に配送する場合に認識する商品もそれぞれ似たところもあるけど異なるもの、
となります。

（もちろん実際とは異なります）

つまり商品は3者の認識が混入しているっぽい、と当たりがつくようになります。
これは開発者だけではなく社内の各方面の方々とコミュニケーションをとりながら、
実際に言葉の齟齬がないか背景や事実確認などを行いながら現状を深掘りします。
社内で共通の言葉を使いながら、領域ごとに多少の背景を汲み取りながら
それぞれの領域で疎通ができるようになればはっきりと境界線が認識できます。
（これは実際に自分でやっています。）

これを認識せずに実際のコードやシステム設計に落とし込むと
データモデルとも大きく乖離してしまうデータベース設計（フラグを多用したり、ライブラリで使いやすい構造になってしまったり）や、
商品という名前を使った巨大ななにかに繋がるコードとなってしまいます。
これはクラスベースの言語でなくても、型演算多用で何者にでもなれてしまう何かになってしまったり
どういう指標で作っていくかはすべて実装者依存になってしまいます。

これでは10年戦えるシステム・仕組みどころか、
実装者が変わった途端に「なぜこう作ったんだろう？」のみが残ってしまい、
開発者の好みなどの粒度でさらに分割したり結合したりとどんどん複雑になってしまいますので、
みなさんも「あれ？」という瞬間があったらぜひ現状の分析をしてみてください。

起点を探す

さて、この記事はここからが本題です。

この境界線はそれぞれの立場や業務などの背景があるのは明確ですが、
それはどのような瞬間から分かれていくのでしょうか。

それは下記のものがキーとなります。

製造以外の分野を引き受けるため、  
ごちそうを商品として登録する運用チーム、営業チームがあり、  
このチームが製造パートナーの商品情報を受け取り、  
サービスが提供する商品として成立するための情報を付与し、  
はじめて喫食者が目にする「ごちそう」となります。

つまり一番最初の事象は、
製造パートナーからシステムを通じて販売したい商品データをもらう・投入してもらう、
という事象を起点に商品に対する認識が少しずつ変化します。

そして商品に対するさまざまな情報が加えられたりすることで「ごちそう」となるわけです。
雑に整理すると下記のようになるでしょう。

製造パートナーが販売したい商品情報を入稿した
スターフェスティバルがサービスで販売するための商品として承認した
商品が「ごちそう」へと生まれ変わった

（多くなるのでこの辺で・・）

もちろんこの事柄以外にもありますが、価値として大きなものはこのようになります。
ではなぜこの事象を取り上げたのでしょうか？

実はこの事象はWebアプリケーション開発の視野と異なる視点でみると
少し違う見方ができます。

これらの事象は製造パートナーとして販売したい商品がどのくらいあり、　販売できなかった商品がどのくらいあったのか、
「ごちそう」と変化した商品がどのくらいあったのか
という事業としても製造パートナーとのやりとりの改善や、
他にもさまざまなエビデンスとなり活用できる価値のあるデータへとつながります。

そして物事の始まりでもある「製造パートナーが販売したい商品情報を入稿した」事象をきっかけに
いろいろな世界へと繋がることが見えてきます。
人によってはまるで小さなビックバンを見つけたかのような感覚にもなるかもしれません。
自分はよく
「今ここでジャンプしたら地球の裏側にはどういうことが起こるかワクワクしながら想像するようなもの」と喩えたりしています。

そしてこの例の事象1つ1つは会社活動やサービス、「ごちそう」など、
それぞれの視点だけで完結するピュアな事象に近いということがわかってきます。
各領域のアプリケーション中ではもっとたくさんの事象がありますが、
会社活動としても重要な事象はどれか、という思考で分類します。

たとえば事象の起点から最終的に「ごちそう」へ変化するまでの道のりを俯瞰することで
物事を把握できるデータ・エビデンスとしても活用できる、
かつ社内外の業務フローからみても、この事象はピュアなまま活用できます。

あくまでたとえばの話ですので、みなさんのアプリケーションや会社活動の中ではどうなのか、
と向き合ってみるとまた違うものが見えてくると思います。
（時系列があることはしっかりと認識しておきましょう）

やがてはこの事象が「あの時どうだったっけ？」という真のデータソースとなり得るもので、
それぞれの立場・領域でこれを元に世界が広がっていく拡張されていくデータの流れとしても捉えることができます。

事象が保管されることでそれぞれの領域に閉じた構造を作ることができるようになります。
お互いを行き来する場合でも相関IDや真のデータソースを介することで
翻訳しあえることも見えてきます。

この事象を更新などで変化をさせずに的確に保管し、
読み込み時にそれぞれの領域の視点を取り入れた構造にすることもできそうです。
(複製しても再度やり直せば良いということにもなります。)

もちろん1つのDBですべてが完結していて
そこで分析やさまざまアプリケーションが動いていて問題がなければ難しくする必要はありません！
あくまで捉え方の例としてください。

どういう仕組みで作っていくか

ここでやっと実装の話に。

事象が発生したタイミングで複数の領域の視点や関心が加わっていくということはわかっています。
この事象が発生したよ、と複数の領域に伝播・指示できれば良さそうです。

もし領域ごとにアプリケーションが存在しているとするとどう取り組んでいくと良いでしょうか？
この場合は社内業務用のアプリケーションに社外用のデータ投入アプリケーション、
そして販売サイトとして作用するアプリケーションとなりますのでアプリケーションが分かれていることになります。
*注意マイクロサービスアーキテクチャではありません。

気をつけなければならないのは、
複数の領域があるためHTTPを挟んだりデータベースへ直接書き込みにいったり、
データ処理のためにBigQueryやS3へ一緒に書き込むぞ！
という選択肢をとると難しい問題と立ち向かわなければなりません。

ここまで述べた中でいくつかの領域があるため、
たとえばシステムが分かれていた場合は3領域に対してデータベースに書き込んだり、
HTTPを挟んで書き込んだりする場合（Web APIなど）は、
どこかの領域で失敗した場合、どこで整合性が担保されるのかという問題があります。

リトライを挟んだとしてどこで3つの領域に書き込まれたよ、と担保できるのでしょうか。

なにかの障害などでどこかの領域に指示ができない場合はどうしていくのでしょうか？

それぞれの領域で構成変更などが起きた・起きる場合は同時に対応しなければならなくなりそうです。

分かれているシステムへ直接データベース書き込みにいくと、
リリースやデータ構造の変更が発生する場合は一緒に行動する必要が出てきます。
加えてどれかの領域の関心が入り込んだテーブルやカラムなどが追加される可能性が高くなっていきます。
どこかで必要だけどどこかで必要ない、という形になっていきます。

このアプローチでは2相コミットが発生するのと、
相手が必ず生きていることを前提とした同期処理ということになってしまいます。

ja.wikipedia.org

つまり指示元が相手先を考慮しなければならなくなります。
お互いに仕様変更がある場合なども大変そうなことも見えてきます。

難しい場合はマイクロサービスアーキテクチャだ！分離だ！と意気込まずに、
まずは結合したアプリケーションとして着手していきましょう。

この例の場合で、各領域に影響を与えず、かつそれぞれの領域独自の視点を入れてもよい、と考えると
メッセージブローカーなどを介するのが良さそうです。

結果整合を選択することになりますが、この例の場合は問題ないという前提にしておきましょう。

事象が起きたことを記憶したい領域がメッセージブローカーに伝えることができれば
指示を受ける側に何かあっても気にする必要はなく、
各領域のタイミングで自由に受け取ることができます。

これにはSQSやMSK、Kinesisなどがありますが
弊社ではMSK（Kafka）を利用しています。
SQSなどのQueueとMSK、Kinesisとの違いなどはまた別の機会にするとして、
ここではKafkaを選択したとしましょう。

*受け取る領域が途中で増えたとしても過去の事象を再送することができるので、再送のために再度指示を投げる必要もありません

メッセージをどうするか？

メッセージブローカーに送信するにはメッセージが必要となります。

メッセージはIDだけを記述して受信側が再度問い合わせる、
という形にすると「あの時どうだったっけ？」ができているつもりでも、
「あの時」が更新された最新のものになってしまうことになります。
必ず事象をそのままスナップショットにしましょう。

肝心のメッセージの送信方法ですが、大まかに2つの方法があります。
1つ目はoutboxパターンを用いてメッセージブローカーに送信、
2つ目はメッセージブローカーに直接送信があります。
（Akka等はこの記事では触れません）

www.infoq.com

outboxパターンを用いる場合は、データベースのトランザクションを使って
送信したいメッセージをテーブルに書き込み、CDC（change data capture）を介してメッセージブローカーに送信されます。

この場合はoutboxとして利用するテーブル以外にも、
送信元のアプリケーションで必要なデータベースに書き込む必要がある場合などにも利用できます。
レガシーアプリケーション改善などにも活用できます。

注意点としてはCDCの対象はテーブル単位となりますので、
複数テーブルの変更をバラバラで受け取ることになりますのですべてを同時に受け取ることは難しくなりますので、
その場合は1つのメッセージで完結するような集約を作る必要があります。
AWSのDynamoDB Streamsなどで用いられている仕組みのようなもの、と捉えてもらえると良いと思います。
WAL（Write Ahead Log）で変更などを検知して吸い上げてメッセージとして送信されるものです。

Debeziumでoutboxパターンをサポートしてくれる機能がありますので、
興味のある方は下記をどうぞ。

debezium.io

すでにスターフェスティバルのアドベントカレンダーに一部がありますので、
こちらも合わせてどうぞ！

zenn.dev

これについてはまたどこかで解説しましょう。

メッセージブローカーに直接送信する場合は、
送信元のアプリケーションでリアルタイムにデータベースへ保存する必要がなく、
メッセージブローカーに送信さえできれば良いという場合に利用します。

この場合は上記にも記述しましたがメッセージブローカーに送ってデータベースにも書き込んで、
といった実装になる場合は2相コミットが発生しますのでなるべく選択することを避けてください。

データベースなどへの書き込みなどはコンシューマーとなるアプリケーションが担当、
もしくはKafka Connectなどを通じて転送して保存できます。

RDBMSやS3、BigQueryやさまざまなデータソースに転送したり、
もちろん前述のCDCとKafka Connectを組み合わせることができます。

なおMSK、Kinesisなどは送信時にトランザクションを利用できますので、確実に届けることができます。

1つの領域で閉じたメッセージ送信などの場合は、アプリケーション側で担保できますが、
本記事の例では複数の領域やデータ処理などにも繋がっていくため、いったんメッセージブローカーを介していきましょう。
（いろんな言語で解説するのが大変なのでGoの例にします）

メッセージを作る

すべての例を記載していくのは大変なので、下記のものだけをやっていきましょう。

製造パートナーが販売したい商品情報を入稿した

特別な知識は排除した簡単な例ですが、おそらく次の構造になるでしょう。

{
  "correlation_id": "11111122223444",
  "store_name": "ytakeキッチン",
  "store_id": 2,
  "product_name": "めちゃくちゃおいしいカレー",
  "wholesale_price": 100,
  "price_including_tax": 108,
  "tax_rate": 1.08,
  "comment": "めちゃくちゃおいしい",
  "type": "new",
  "created_at":"2022-12-12T00:00:00+09:00"
}

だいぶ省略していますが税率が複数ある場合はこのままでは対応できませんので、税率を含めて構造化等を。
データベースに直接突っ込むものではありませんので、
実際に永続化するときは商品と価格などは分けておきましょう。

package message

import "time"

type EventType string

const (
    New    EventType = "new"
    UPDATE EventType = "update"
)

// Product 商品に関するメッセージを表現したもの
type Product struct {
    // CorrelationID 相関ID
    CorrelationID string `json:"correlation_id"`
    // StoreName 製造パートナー名
    StoreName string `json:"store_name"`
    // StoreID 製造パートナーID
    StoreID int `json:"store_id"`
    // ProductName 商品名
    ProductName string `json:"product_name"`
    // WholesalePrice　卸値　税抜
    WholesalePrice int `json:"wholesale_price"`
    // PriceIncludingTax 卸値 税込
    PriceIncludingTax int `json:"price_including_tax"`
    // TaxRate 税率
    TaxRate float64 `json:"tax_rate"`
    // Comment 何かあれば
    Comment string `json:"comment"`
    // Type データ入稿が新規か更新かなどなど
    Type EventType `json:"type"`
    // CreatedAt 事象の発生日時
    CreatedAt time.Time `json:"created_at"`
}

良さそうに見えますね。

待ってください？！

たとえばデータ投入で少し変わってきた場合にどうやって構造が変更されたことを担保するのでしょうか？
型指定があまり得意ではない（型指定しているように見えてもそれはあくまでコード上だけだったりも然り）場合はどうやったら？

という場合には
Apache AvroやProtocol Buffersを利用するといいでしょう。

これらを使うことで各アプリケーションやデータ処理などで定義を共有できます。
下記はProtocol Buffersの例です。

syntax = "proto3";

package product;
import "google/protobuf/timestamp.proto";

option java_package = "com.github.ytake.example.protobuf";
option go_package = "github.com/ytake/example/pb";

message RegistrationAction {
  uint64 correlationId = 1;
  string storeName = 2;
  uint32 storeId = 3;
  string productName = 4;
  uint32 wholesalePrice = 5;
  uint32 priceIncludingTax = 6;
  float taxRate = 7;
  string comment = 8;
  enum EventType {
    NEW = 0;
    UPDATED = 1;
  }
  EventType event = 9;
  google.protobuf.Timestamp created = 10;
}

言語に対応したoptionなどもありますので必要に応じて足します。
AWS DMSやDebeziumなどでoutboxとしてこれらを利用する場合は下記を参照してみるといいでしょう。

medium.com

さてとこれで。。

おっと、まだ考慮しておくことがあります。
これまで述べた長い背景の中で事象には順序があるらしい、とわかっています。
たとえばこの事象ごとにQueueやTopicを分けて送信してしまうと、
送った側は時系列順に送ったつもりでも受け取り側はそうなるとは限りません。

どこかでスタックしてしまうと、時系列は想定通りにはならず簡単に順番が入れ替わってしまいます。
スーパーのレジを想像してみてください。
並んでる人が少ない列で待っていると、「あれ、あっちの方が多かったはずなのにもう捌けてる・・」
あの現象が起こります。
事象の分析や時系列を理解してメッセージ設計に含めましょう。

今回の例であれば最終的には付加情報をつけて「ごちそう」になるわけですから、
前の状態がなくとも突然「ごちそう」のメッセージが流れてきても問題ないかどうか、
時系列の判断などが可能かどうか、スナップショットと不整合が起きないかどうか、
結果整合となりますのでそちらで担保が可能で、
頭から読み出して再処理しても問題ないかどうか、などなどがあります。

上記のprotoファイルに、たとえば付加情報を含めてスナップショットとする、などなど
Protocol BuffersのOptional なども理解しておきましょう。

このあたりの設計について触れるとどんどん長くなってしまうので、今回は割愛します。
機会があればこれはこれでどこかで・・・

protoファイルを使ってGoのコードを吐き出してあげます。

# Goのプロジェクトルートから実行して./pbに吐き出すとしたら
$  protoc --go_out=./pb --go_opt=paths=source_relative ./product.proto

簡単なサンプルですが、下記のようになります。

import (
    "errors"
    "github.com/confluentinc/confluent-kafka-go/kafka"
    "github.com/ytake/example/pb"
    "golang.org/x/xerrors"
    "google.golang.org/protobuf/proto"
    "google.golang.org/protobuf/types/known/timestamppb"
    "strconv"
)

// 抜粋
    ra := pb.RegistrationAction{
        CorrelationID:     11111122223444,
        StoreName:         "ytakeキッチン",
        StoreId:           2,
        ProductName:       "めちゃくちゃおいしいカレー",
        WholesalePrice:    100,
        PriceIncludingTax: 108,
        TaxRate: 1.08,
        Comment:           "めちゃくちゃおいしい",
        Event:             pb.RegistrationAction_NEW,
        Created:           timestamppb.Now(),
    }
    tn := "topic-name"
    by, err := proto.Marshal(&ra)
    if err != nil {
        return xerrors.Errorf("error: %w", ErrProtobufMarshal)
    }
    deliveryChan := make(chan kafka.Event)
    err = p.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{
            Topic:     &tn,
            Partition: kafka.PartitionAny,
        },
        Value: by,
        Key:   []byte(strconv.Itoa(int(ra.CorrelationID))), // パーティションKey
        Headers:        []kafka.Header{{Key: "適切なヘッダーKey", Value: []byte("header values are binary")}},
    }, deliveryChan)
// 以下略
// close等していませんのでこれをコピペしても動きませんので注意

これで良さそうです。

いや、まだあります。

メッセージの定義は時の流れとともに変更されていきます。
開発していくなかで要件が変わったり、設計見直しなどがありますのでずっとこのままというわけにはいきません。

そんな時にオススメなのはSchema Registryです。

docs.confluent.io

AWSではGlue Schema Registryがあります。
（GCPは利用していないのでわかりませんがCloud Pub/Subとか？）

もちろんすべての変更に完璧に応えるというわけではなく、
どのように互換性を担保するかを指定しなければなりません。
互換性チェックのタイプとしては大まかに下記の通りです。

後方互換性
前方互換性
完全互換性
互換性チェックなし

コンシューマーから先にアップグレードするかプロデューサーから先か、など
ユースケースによって選ぶことができます。
詳細を解説すると本1冊分くらいになりますので、下記を参照してください。

docs.confluent.io

互換性の指示についてはSchema RegistryのREST APIを利用するか、
クライアントで指定できます。
Schema Registryを介してKafkaへ送信する場合は下記のように利用できます。

// 抜粋
    p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "bootstrapServers"})
    if err != nil {
        return err
    }
    client, err := schemaregistry.NewClient(schemaregistry.NewConfig(url))
    if err != nil {
        return xerrors.Errorf("Failed to create schema registry client: %w", ErrSchemaRegistryClient)
    }

    /* subjectを指定して互換性を変更する場合は下記のように利用できます
    *
    *_, err = client.UpdateCompatibility("subject名", schemaregistry.Forward)
    *if err != nil {
    *  return xerrors.Errorf("Failed to change compatibility: %w", ErrSchemaRegistryCompatibility)
    *}
    */

    ser, err := protobuf.NewSerializer(client, serde.ValueSerde, protobuf.NewSerializerConfig())

    ra := pb.RegistrationAction{
        CorrelationID:     11111122223444,
        StoreName:         "ytakeキッチン",
        StoreId:           2,
        ProductName:       "めちゃくちゃおいしいカレー",
        WholesalePrice:    100,
        PriceIncludingTax: 108,
        TaxRate: 1.08,
        Comment:           "めちゃくちゃおいしい",
        Event:             pb.RegistrationAction_NEW,
        Created:           timestamppb.Now(),
    }
    tn := "topic-name"
    payload, err := ser.Serialize(tn, &ra)
    if err != nil {
        return xerrors.Errorf("error: %w", ErrProtobufMarshal)
    }
    deliveryChan := make(chan kafka.Event)
    err = p.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{
            Topic:     &tn,
            Partition: kafka.PartitionAny,
        },
        Value:   payload,
        Key:     []byte(strconv.Itoa(int(ra.CorrelationID))),
        Headers: []kafka.Header{{Key: "適切なヘッダーKey", Value: []byte("header values are binary")}},
    }, deliveryChan)
    // 以下略