Rustのtonic（gRPC）でDDD（ドメイン駆動設計）構成のバックエンドAPIを開発する方法まとめ

FROM rust:1.87

WORKDIR /app

COPY . .

# インストール可能なパッケージ一覧の更新
RUN apt-get update && \
    # パッケージのインストール
    apt-get install -y \
      protobuf-compiler \
      curl

# gRPCのドキュメント生成やバリデーション用にgoをインストール
# goのバージョンを変更したい場合は以下リンク先のページで必要情報をご確認下さい。
# https://go.dev/dl/
ENV GO_VERSION=1.24.4
ENV GO_SHA256=d5501ee5aca0f258d5fe9bfaed401958445014495dc115f202d43d5210b45241
ENV GO_OS=linux
ENV GO_ARCH=arm64
ENV PATH="/usr/local/go/bin:/root/go/bin:${PATH}"

RUN curl -fsSLO "https://go.dev/dl/go${GO_VERSION}.${GO_OS}-${GO_ARCH}.tar.gz" && \
    echo "${GO_SHA256} go${GO_VERSION}.${GO_OS}-${GO_ARCH}.tar.gz" | sha256sum -c - && \
    tar -C /usr/local -xzf "go${GO_VERSION}.${GO_OS}-${GO_ARCH}.tar.gz" && \
    rm "go${GO_VERSION}.${GO_OS}-${GO_ARCH}.tar.gz" && \
    rm -rf /var/lib/apt/lists/*

# gRPCのドキュメント生成用にgoのライブラリをインストール
RUN go install github.com/pseudomuto/protoc-gen-doc/cmd/protoc-gen-doc@v1.5.1

# バリデーション用にgoのライブラリをインストール
RUN go install github.com/envoyproxy/protoc-gen-validate@v1.2.1

# ホットリロード用のライブラリをインストール
RUN cargo install cargo-watch

# Rust用のリンターをインストール
RUN rustup component add clippy

# Rust用のフォーマッターをインストール
RUN rustup component add rustfmt

use chrono::TimeZone;
use std::io::Write;

// ロガー用のトレイト
use crate::application::usecase::logger::logger_trait::LoggerTrait;

// 共通コンテキスト
use crate::application::usecase::context::context_request::ContextRequest;

// ロガーの構造体
#[derive(Clone, Debug)]
pub struct Logger {}

impl Logger {
    // ロガーの初期化処理
    pub fn init() {
        // 日本時間を取得
        let jst = chrono::offset::FixedOffset::east_opt(9 * 3600)
            .unwrap()
            .from_utc_datetime(&chrono::Utc::now().naive_utc());

        // カスタムロガーの初期化
        env_logger::builder()
            .format(move |buf, record| {
                writeln!(
                    buf,
                    "{} {} {}",
                    jst.format("%Y-%m-%d %H:%M:%S"),
                    record.level(),
                    record.args()
                )
            })
            .init();
    }

    // インスタンス生成
    pub fn new() -> Self {
        Logger {}
    }

    // コンテキストからリクエスト情報取得
    fn get_req_info_from_ctx(ctx: &ContextRequest) -> String {
        format!("request_id={} uri={}", ctx.request_id, ctx.uri)
    }
}

#[async_trait::async_trait]
impl LoggerTrait for Logger {
    fn info(&self, ctx: &ContextRequest, msg: &str) {
        let req_info = Logger::get_req_info_from_ctx(ctx);
        log::info!("[{}] {}", req_info, msg);
    }

    fn warn(&self, ctx: &ContextRequest, msg: &str) {
        let req_info = Logger::get_req_info_from_ctx(ctx);
        log::warn!("[{}] {}", req_info, msg);
    }

    fn error(&self, ctx: &ContextRequest, msg: &str) {
        let req_info = Logger::get_req_info_from_ctx(ctx);
        log::error!("[{}] {}", req_info, msg);
    }
}

use hyper::header::HeaderValue;
use std::{
    pin::Pin,
    task::{Context, Poll},
};
use tonic::server::NamedService;
use tower::{Layer, Service};
use uuid::Uuid;

// 共通コンテキスト
use crate::application::usecase::context::context_request::ContextRequest;

// ロガー
use crate::application::usecase::logger::logger_trait::LoggerTrait;
use crate::infrastructure::logger::logger_log::Logger;

// リクエスト用のミドルウェア
#[derive(Clone)]
pub struct RequestMiddleware<S> {
    inner: S,
}

impl<S> RequestMiddleware<S> {
    fn new(inner: S) -> Self {
        RequestMiddleware { inner }
    }
}

// リクエスト用のミドルウェアにServiceトレイトを実装
impl<B, S> Service<hyper::Request<B>> for RequestMiddleware<S>
where
    S: Service<hyper::Request<B>> + Clone + Send + 'static,
    S::Future: Send,
    B: Send + 'static,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut req: hyper::Request<B>) -> Self::Future {
        // リクエストからuriを取得
        let uri_path = req.uri().path();

        // uriをリクエストヘッダーに設定
        let uri_string = match HeaderValue::from_str(uri_path) {
            Ok(value) => {
                req.headers_mut().insert("x-uri", value.clone());
                value.to_str().unwrap().to_string()
            }
            Err(_) => {
                // エラーの場合は「-」を設定
                let header_value = HeaderValue::from_str("-").unwrap();
                req.headers_mut().insert("x-uri", header_value.clone());
                header_value.to_str().unwrap().to_string()
            }
        };

        // request-idをリクエストヘッダーに設定
        let uuid = Uuid::new_v4().to_string();
        req.headers_mut()
            .insert("x-request-id", uuid.clone().parse().unwrap());

        // リクエスト開始ログ
        let ctx = ContextRequest {
            request_id: uuid,
            uri: uri_string,
        };
        let logger = Logger::new();
        logger.info(&ctx, "Execute gRPC request !!");

        // 非同期処理のためself.innerをコピー
        let mut inner = self.inner.clone();

        // 非同期処理
        Box::pin(async move { inner.call(req).await })
    }
}

// リクエスト用のミドルウェアにtonic::server::NamedServiceを実装
impl<S> NamedService for RequestMiddleware<S>
where
    S: NamedService,
{
    const NAME: &'static str = S::NAME;
}

// リクエスト用のミドルウェアのレイヤー
#[derive(Clone)]
pub struct RequestMiddlewareLayer;

// リクエスト用のミドルウェアのレイヤーにLayerトレイトの実装
impl<S> Layer<S> for RequestMiddlewareLayer {
    type Service = RequestMiddleware<S>;

    fn layer(&self, inner: S) -> Self::Service {
        RequestMiddleware::new(inner)
    }
}

use tonic::{Code, Request, Status};

#[allow(clippy::result_large_err)]
pub fn auth_interceptor(req: tonic::Request<()>) -> Result<Request<()>, Status> {
    // メタデータからx-uriを取得
    let uri = match req.metadata().get("x-uri") {
        Some(value) => value.to_str().unwrap_or_default(),
        None => "-",
    };

    // 対象のuriの場合はスキップ
    let skip_uri: &[&str] = &[
        // "/chat.ChatService/Bidirectional",
    ];
    if skip_uri.contains(&uri) {
        return Ok(req);
    }

    // リクエストヘッダーからトークン取得
    let token = match req.metadata().get("authorization") {
        Some(value) => {
            let bearer_token = value.to_str().unwrap_or_default();
            bearer_token.trim_start_matches("Bearer ")
        }
        None => "",
    };

    // 認証トークンが設定されていない場合はエラー
    if token.is_empty() {
        let status = Status::new(
            Code::InvalidArgument,
            "認証用トークンが設定されていません。",
        );
        return Err(status);
    }

    // TODO: 認証チェック処理を追加

    // 戻り値
    Ok(req)
}

use serde::{Deserialize, Serialize};

// chatモデル
#[derive(Clone, Serialize, Deserialize, Debug)]
pub struct Chat {
    pub input_text: String,
}

impl Chat {
    // 新規作成
    pub fn new(input_text: String) -> Self {
        Self { input_text }
    }

    // 大文字変換
    pub fn text_to_upper(&self) -> String {
        self.input_text.to_uppercase()
    }
}

#[cfg(test)]
mod tests {
    use crate::domain::chat::chat_model::Chat;

    #[test]
    fn test_new_chat() {
        // textパラメータ
        let text = "hello".to_string();

        // テスト実行
        let chat = Chat::new(text.clone());

        // 検証
        assert_eq!(chat.input_text, text);
    }

    #[test]
    fn test_text_to_upper() {
        // textパラメータ
        let text = "hello".to_string();

        // テスト実行
        let chat = Chat::new(text.clone());

        // 検証
        assert_eq!(chat.text_to_upper(), "HELLO");
    }
}

pub mod chat_model;

// テストコード用のモジュール
pub mod chat_model_test;

pub mod chat;
pub mod error;

// tonic
use tonic::{
    Request, Response, Status, Streaming,
    metadata::{Ascii, MetadataValue},
};

// tokio
use tokio::{
    spawn,
    sync::mpsc,
    time::{Duration, sleep},
};
use tokio_stream::{StreamExt, wrappers::ReceiverStream};

// 変換用
use std::str::FromStr;

// バリデーション用のトレイト
use prost_validate::Validator;

// Arc（ヒープ上に確保されたある値の所有権を、複数のスレッド間で安全に共有するためのスマートポインタ）
use std::sync::Arc;

// 共通コンテキスト
use crate::application::usecase::context::context_request::new_context_request;

// ロガー
use crate::application::usecase::logger::logger_trait::LoggerTrait;

// Chatドメイン
use crate::domain::chat::chat_model::Chat;

// ビルドしたprotoファイルのインポート
pub mod chat_proto {
    // protoファイルに定義したpakage名を指定
    tonic::include_proto!("chat");
}

// ストリーミング用の型定義
type BidirectionalStream = ReceiverStream<Result<chat_proto::TextOutput, Status>>;

// ユースケース用のトレイト（モック化もできるように定義）
#[mockall::automock]
#[async_trait::async_trait]
pub trait ChatBidirectionalUsecaseTrait {
    async fn exec(
        &self,
        request: Request<Streaming<chat_proto::TextInput>>,
    ) -> Result<Response<BidirectionalStream>, Status>;
}

// 実行するユースケースの構造体
#[derive(Clone)]
pub struct ChatBidirectionalUsecase {
    pub logger: Arc<dyn LoggerTrait + 'static>,
}

impl ChatBidirectionalUsecase {
    pub fn new(logger: Arc<dyn LoggerTrait + 'static>) -> Self {
        ChatBidirectionalUsecase { logger }
    }
}

#[async_trait::async_trait]
impl ChatBidirectionalUsecaseTrait for ChatBidirectionalUsecase {
    async fn exec(
        &self,
        request: Request<Streaming<chat_proto::TextInput>>,
    ) -> Result<Response<BidirectionalStream>, Status> {
        // コンテキスト設定
        let ctx = new_context_request(&request);

        // トレーラー用
        let x_request_id = MetadataValue::<Ascii>::from_str(ctx.request_id.as_str()).expect("-");

        // リクエストからストリームを取り出す
        let mut stream = request.into_inner();

        // mpsc (multi-producer, single-consumer) チャンネルの作成
        // サーバーはこのチャンネルにデータ送信し、その後クライアントにストリーミングする
        // バッファサイズは適宜調整が必要
        let (tx, rx) = mpsc::channel(1);

        // ロガー
        let logger = self.logger.clone();

        spawn(async move {
            logger.info(&ctx, "Start Bidirectional Stream !!");
            while let Some(result) = stream.next().await {
                match result {
                    Ok(req) => {
                        // バリデーションチェック
                        match req.validate() {
                            Ok(_) => {}
                            Err(e) => {
                                let msg = format!("バリデーションエラー: {}", e);
                                logger.error(&ctx, &msg);
                                let mut status = Status::invalid_argument(msg);
                                status
                                    .metadata_mut()
                                    .insert("x-request-id", x_request_id.clone());
                                let _ = tx.send(Err(status)).await;
                                break;
                            }
                        }

                        // Chatモデルの設定
                        let chat = Chat::new(req.text);

                        // レスポンスの設定
                        let res = chat_proto::TextOutput {
                            text: chat.text_to_upper(),
                        };

                        // Okでラップしたレスポンスをtxで送信
                        if let Err(e) = tx.send(Ok(res)).await {
                            // クライアントの接続切れなどでエラーの場合
                            let msg = format!("Failed to send data: {:?}", e);
                            logger.error(&ctx, &msg);
                            let mut status = Status::invalid_argument(msg);
                            status
                                .metadata_mut()
                                .insert("x-request-id", x_request_id.clone());
                            let _ = tx.send(Err(status)).await;
                            break;
                        }

                        // 1秒間待機処理
                        sleep(Duration::from_secs(1)).await;
                    }
                    Err(e) => {
                        // クライアントの接続切れなどでエラーの場合
                        let msg = format!("Failed to send data: {:?}", e);
                        logger.error(&ctx, &msg);
                        let mut status = Status::invalid_argument(msg);
                        status
                            .metadata_mut()
                            .insert("x-request-id", x_request_id.clone());
                        let _ = tx.send(Err(status)).await;
                        break;
                    }
                }
            }

            // トレーラーの設定
            let mut status = Status::ok("Stream finished successfully");
            status
                .metadata_mut()
                .insert("x-request-id", x_request_id.clone());

            // Errでラップしたステータスを送信
            if let Err(e) = tx.send(Err(status)).await {
                // クライアントの接続切れなどでエラーの場合
                let msg = format!("Failed to send data: {:?}", e);
                logger.error(&ctx, &msg);
                let mut status = Status::invalid_argument(msg);
                status.metadata_mut().insert("x-request-id", x_request_id);
                let _ = tx.send(Err(status)).await;
            }

            logger.info(&ctx, "Finish Bidirectional Stream !!");
        });

        Ok(Response::new(ReceiverStream::new(rx)))
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use tokio_stream::StreamExt;
    use tonic_mock::streaming_request;

    // ロガーのモック
    use crate::application::usecase::logger::logger_trait::MockLoggerTrait;

    // ユースケース
    use crate::application::usecase::chat::chat_bidirectional::{
        ChatBidirectionalUsecase, ChatBidirectionalUsecaseTrait, chat_proto,
    };

    #[tokio::test]
    async fn test_exec_success() {
        // ロガーのモック化
        let mut mock_logger = MockLoggerTrait::new();
        mock_logger.expect_info().returning(|_, _| ());

        // ユースケースのインスタンス化
        let chat_bidirectional_usecase = ChatBidirectionalUsecase {
            logger: Arc::new(mock_logger),
        };

        // 送信するメッセージリスト作成
        let message_lists = vec![chat_proto::TextInput {
            text: "hello".to_string(),
        }];

        // tonic-mockを使ってストリーミングリクエストを作成
        let request = streaming_request(message_lists);

        // テスト実行
        let res = chat_bidirectional_usecase.exec(request).await;

        // 検証
        assert!(res.is_ok());

        // ストリームの検証
        let mut stream = res.unwrap().into_inner();
        let mut i = 0;
        let outputs = vec![chat_proto::TextOutput {
            text: "HELLO".to_string(),
        }];
        while let Some(result) = stream.next().await {
            match result {
                Ok(res) => {
                    let msg = outputs[i].text.clone();
                    assert_eq!(res.text, msg);
                }
                Err(_) => {}
            }
            i += 1;
        }
    }

    #[tokio::test]
    async fn test_exec_validation_error() {
        // ロガーのモック化
        let mut mock_logger = MockLoggerTrait::new();
        mock_logger.expect_info().returning(|_, _| ());
        mock_logger.expect_error().returning(|_, _| ());

        // ユースケースのインスタンス化
        let chat_bidirectional_usecase = ChatBidirectionalUsecase {
            logger: Arc::new(mock_logger),
        };

        // 送信するメッセージリスト作成
        let message_lists = vec![chat_proto::TextInput {
            text: "".to_string(),
        }];

        // tonic-mockを使ってストリーミングリクエストを作成
        let request = streaming_request(message_lists);

        // テスト実行
        let res = chat_bidirectional_usecase.exec(request).await;

        // 検証
        assert!(res.is_ok());

        // ストリームの検証
        let mut stream = res.unwrap().into_inner();
        while let Some(result) = stream.next().await {
            match result {
                Ok(_) => {}
                Err(err) => {
                    assert_eq!(err.code(), tonic::Code::InvalidArgument);
                    assert!(err.message().contains("バリデーションエラー"));
                    break;
                }
            }
        }
    }
}

use std::sync::Arc;

// ロガー
use crate::infrastructure::logger::logger_log::Logger;

// ユースケース
use crate::application::usecase::chat::chat_bidirectional::ChatBidirectionalUsecase;

// Chatユースケース
#[derive(Clone)]
pub struct ChatUsecase {
    pub chat_bidirectional: ChatBidirectionalUsecase,
}

// アプリケーション全体で共有する状態（DIコンテナ）
#[derive(Clone)]
pub struct AppState {
    pub chat_usecase: ChatUsecase,
}

impl AppState {
    pub async fn new() -> Self {
        // ロガー設定
        let logger = Arc::new(Logger::new());

        // Chatユースケースのインスタンス化とまとめ
        let chat_bidirectional_usecase = ChatBidirectionalUsecase::new(logger.clone());
        let chat_usecase = ChatUsecase {
            chat_bidirectional: chat_bidirectional_usecase,
        };

        // 戻り値の設定
        Self { chat_usecase }
    }
}

// tonic
use tonic::{Request, Response, Status, Streaming};

// tokio
use tokio_stream::wrappers::ReceiverStream;

// Arc（ヒープ上に確保されたある値の所有権を、複数のスレッド間で安全に共有するためのスマートポインタ）
use std::sync::Arc;

// レジストリ
use crate::registry::registry_settings::AppState;

// Chatユースケース
use crate::application::usecase::chat::chat_bidirectional::{
    ChatBidirectionalUsecaseTrait, chat_proto,
};

// 構造体定義
pub struct ChatServer {
    state: Arc<AppState>,
}

// protoファイルの関数の実装をメソッド定義
#[tonic::async_trait]
impl chat_proto::chat_service_server::ChatService for ChatServer {
    // 双方向ストリーミングの追加（typeの定義必須）
    // ※protoファイルで定義した名称+Streamという型の定義が必要になる
    type BidirectionalStream = ReceiverStream<Result<chat_proto::TextOutput, Status>>;

    async fn bidirectional(
        &self,
        request: Request<Streaming<chat_proto::TextInput>>,
    ) -> Result<Response<Self::BidirectionalStream>, Status> {
        // ユースケースの実行
        self.state
            .chat_usecase
            .chat_bidirectional
            .exec(request)
            .await
    }
}

// ルーターに設定するサーバー定義を返す関数
pub fn get_chat_server(
    state: Arc<AppState>,
) -> chat_proto::chat_service_server::ChatServiceServer<ChatServer> {
    let chat_server = ChatServer { state };
    chat_proto::chat_service_server::ChatServiceServer::new(chat_server)
}

#[cfg(test)]
mod tests {
    use tokio_stream::{StreamExt, iter};
    use tonic::Request;

    // コンフィグ設定
    use crate::config::config_settings::get_config;

    use crate::application::usecase::chat::chat_bidirectional::chat_proto::{
        TextInput, TextOutput, chat_service_client::ChatServiceClient,
    };

    #[tokio::test]
    async fn bidirectional_should_return_succeed() {
        let config = get_config();

        // サーバーのインスタンス作成
        let mut client =
            ChatServiceClient::connect(format!("http://localhost:{}", config.grpc_port))
                .await
                .unwrap();

        // 送信するメッセージリスト作成
        let message_lists = vec![TextInput {
            text: "hello".to_string(),
        }];

        // ストリームの作成
        let mut request = Request::new(iter(message_lists.clone()));

        // メタデータにBearerトークンを追加
        request
            .metadata_mut()
            .insert("authorization", "Bearer token".parse().unwrap());

        // テストの実行
        let res = client.bidirectional(request).await;

        // 検証
        assert!(res.is_ok());

        // ストリームの検証
        let mut stream = res.unwrap().into_inner();
        let mut i = 0;
        let outputs = vec![TextOutput {
            text: "HELLO".to_string(),
        }];
        while let Some(result) = stream.next().await {
            match result {
                Ok(res) => {
                    let msg = outputs[i].text.clone();
                    assert_eq!(res.text, msg);
                }
                Err(_) => {}
            }
            i += 1;
        }
    }

    #[tokio::test]
    async fn bidirectional_should_return_auth_error() {
        let config = get_config();

        // サーバーのインスタンス作成
        let mut client =
            ChatServiceClient::connect(format!("http://localhost:{}", config.grpc_port))
                .await
                .unwrap();

        // 送信するメッセージリスト作成
        let message_lists = vec![TextInput {
            text: "hello".to_string(),
        }];

        // ストリームの作成
        let request = Request::new(iter(message_lists.clone()));

        // テストの実行
        let res = client.bidirectional(request).await;

        // 検証
        assert!(res.is_err());
        let err = res.unwrap_err();
        assert_eq!(err.code(), tonic::Code::InvalidArgument);
        assert!(
            err.message()
                .contains("認証用トークンが設定されていません。")
        );
    }
}

pub mod chat_server;

// テストコード用のモジュール
pub mod chat_server_test;

// tonic
use tonic::{service::InterceptorLayer, transport::Server};
use tonic_reflection::server::Builder as ReflectionBuilder;

// tower
use tower::ServiceBuilder;

// Arc（ヒープ上に確保されたある値の所有権を、複数のスレッド間で安全に共有するためのスマートポインタ）
use std::sync::Arc;

// レジストリ
use crate::registry::registry_settings::AppState;

// コンフィング設定
use crate::config::config_settings::get_config;

// ミドルウェア
use crate::presentation::middleware::request_middleware::RequestMiddlewareLayer;

// インターセプター
use crate::presentation::interceptor::auth_interceptor::auth_interceptor;

// gRPCサーバーのサービス
use crate::presentation::server::grpc::chat::chat_server::get_chat_server;

// protoファイル
pub mod proto {
    pub const CHAT_DESCRIPTOR: &[u8] = tonic::include_file_descriptor_set!("chat_descriptor");
}

pub async fn router(state: Arc<AppState>) -> Result<(), Box<dyn std::error::Error>> {
    // 環境変数取得
    let config = get_config();

    // アドレス設定
    let grpc_port = config.grpc_port;
    let addr = format!("[::]:{}", grpc_port).parse()?;

    // サーバー設定
    let chat_server = get_chat_server(state.clone());

    // サーバーリフレクション設定（旧ツールではv1alphaを使う）
    let reflection_service_v1 = ReflectionBuilder::configure()
        .register_encoded_file_descriptor_set(proto::CHAT_DESCRIPTOR)
        .build_v1()?;

    let reflection_service_v1alpha = ReflectionBuilder::configure()
        .register_encoded_file_descriptor_set(proto::CHAT_DESCRIPTOR)
        .build_v1alpha()?;

    // サービス設定
    let service = ServiceBuilder::new()
        .layer(RequestMiddlewareLayer)
        .layer(InterceptorLayer::new(auth_interceptor))
        .service(chat_server);

    // サーバー起動
    Server::builder()
        .add_service(reflection_service_v1)
        .add_service(reflection_service_v1alpha)
        .add_service(service)
        .serve(addr)
        .await?;

    Ok(())
}

// Arc（ヒープ上に確保されたある値の所有権を、複数のスレッド間で安全に共有するためのスマートポインタ）
use std::sync::Arc;

// モジュールのインポート
mod application;
mod config;
mod domain;
mod infrastructure;
mod presentation;
mod registry;

// コンフィグ設定
use crate::config::config_settings::get_config;

// ルーター設定
use crate::presentation::router::router_settings::router;

// ロガー設定
use crate::infrastructure::logger::logger_log::Logger;

// レジストリ設定
use crate::registry::registry_settings::AppState;

#[tokio::main]
async fn main() {
    // 環境変数取得
    let config = get_config();

    // ロガーの初期化
    Logger::init();

    // サーバー起動のログ出力
    log::info!("[ENV={}] Start rust_grpc_domain !!", config.env);

    // サーバー起動
    let state = Arc::new(AppState::new().await);
    router(state).await.unwrap();
}