cqrs/
cqrs.rs

1//! CQRS: separate a command that adjusts an order total from a query that reads it.
2//! CQRS: 注文金額を増やす命令と、金額を読む質問の責務を分ける。
3//!
4//! Run: cargo run -p ddd-toolkit --example cqrs --locked
5//! Read main → AddToOrder::handle → SaveThenDispatch → Total, then support/mod.rs.
6//! main → 命令の処理 → 保存/配信の順序 → クエリ → 共通アダプターの順に読む。
7//! Expected: reject zero / totals above 5000, then query a total of 4200.
8//! 0の加算と上限超過を拒否し、200を加算して4200を参照する。
9//! Commands and queries share a repository; support/mod.rs defines the Order model.
10//! 命令とクエリはRepositoryを共有する。Orderの定義はsupport/mod.rsを参照。
11mod support;
12
13use ddd_toolkit::{
14    AggregateRoot as _, CommandHandler, LifecycleCallback, LifecycleCompletionError,
15    LifecycleContext, LifecycleOperationError, Policy, QueryHandler, Repository, Specification,
16};
17use std::convert::Infallible;
18use support::{Order, Placed, Publisher, Store};
19
20// A reusable yes/no condition. Adapter failure and an unsatisfied rule differ.
21// 再利用可能な真偽条件。条件不成立と評価自体の失敗は別。
22struct Positive;
23impl Specification<u32> for Positive {
24    type Error = Infallible;
25    fn is_satisfied_by(&self, amount: &u32) -> Result<bool, Infallible> {
26        Ok(*amount > 0)
27    }
28}
29// A constraint that can explain rejection; it does not persist anything.
30// 拒否理由を返せる制約。Policy 自体は保存を行わない。
31struct Limit;
32impl Policy<u32> for Limit {
33    type Error = Error;
34    fn check(&self, total: &u32) -> Result<(), Error> {
35        if *total <= 5000 {
36            Ok(())
37        } else {
38            Err(Error::Rejected)
39        }
40    }
41}
42
43#[derive(Debug, PartialEq)]
44enum Error {
45    NotFound,
46    Rejected,
47    Operation(LifecycleOperationError<&'static str>),
48    Completion(LifecycleCompletionError),
49}
50impl From<LifecycleOperationError<Infallible>> for Error {
51    fn from(error: LifecycleOperationError<Infallible>) -> Self {
52        match error {
53            LifecycleOperationError::Adapter(never) => match never {},
54            LifecycleOperationError::AlreadySaved => {
55                Self::Operation(LifecycleOperationError::AlreadySaved)
56            }
57            LifecycleOperationError::AlreadyDispatched => {
58                Self::Operation(LifecycleOperationError::AlreadyDispatched)
59            }
60        }
61    }
62}
63impl From<LifecycleOperationError<&'static str>> for Error {
64    fn from(error: LifecycleOperationError<&'static str>) -> Self {
65        Self::Operation(error)
66    }
67}
68impl From<LifecycleCompletionError> for Error {
69    fn from(error: LifecycleCompletionError) -> Self {
70        Self::Completion(error)
71    }
72}
73
74struct AddToOrder {
75    store: Store,
76    publisher: Publisher,
77}
78impl CommandHandler for AddToOrder {
79    type Command = u32;
80    type Aggregate = Order;
81    type Repository = Store;
82    type Publisher = Publisher;
83    type Output = ();
84    type Error = Error;
85    fn handle<F>(&mut self, id: &u64, amount: u32, callback: F) -> Result<(), Error>
86    where
87        F: for<'a> LifecycleCallback<'a, Order, Store, Publisher, Output = (), Error = Error>,
88    {
89        let mut order = self.store.find_by_id(id).unwrap().ok_or(Error::NotFound)?;
90        if !Positive.is_satisfied_by(&amount).unwrap() {
91            return Err(Error::Rejected);
92        }
93        let total = order.total.checked_add(amount).ok_or(Error::Rejected)?;
94        Limit.check(&total)?;
95        order.total = total;
96        order.record_event(Placed(order.id));
97        let mut context = LifecycleContext::new(&mut order, &mut self.store, &mut self.publisher);
98        callback.call(&mut context)?;
99        context.complete()?;
100        Ok(())
101    }
102}
103// Application orchestration, not a database transaction: dispatch can fail AFTER save.
104// アプリケーションの手順。トランザクションではなく、保存成功後に配信だけ失敗し得る。
105struct SaveThenDispatch;
106impl<'storage> LifecycleCallback<'storage, Order, Store, Publisher> for SaveThenDispatch {
107    type Output = ();
108    type Error = Error;
109    fn call<'callback>(
110        self,
111        context: &'callback mut LifecycleContext<'storage, Order, Store, Publisher>,
112    ) -> Result<(), Error>
113    where
114        'storage: 'callback,
115    {
116        context.save()?;
117        context.dispatch()?;
118        Ok(())
119    }
120}
121struct Total<'a>(&'a Store);
122impl QueryHandler for Total<'_> {
123    type Query = u64;
124    type Result = u32;
125    type Error = Error;
126    fn handle(&self, id: u64) -> Result<u32, Error> {
127        self.0
128            .find_by_id(&id)
129            .unwrap()
130            .map(|order| order.total)
131            .ok_or(Error::NotFound)
132    }
133}
134
135fn main() {
136    let mut handler = AddToOrder {
137        store: Store::default(),
138        publisher: Publisher::default(),
139    };
140    handler
141        .store
142        .save(&Order {
143            id: 1,
144            total: 4000,
145            events: vec![],
146        })
147        .unwrap();
148    assert_eq!(
149        handler.handle(&1, 0, SaveThenDispatch),
150        Err(Error::Rejected)
151    );
152    assert_eq!(
153        handler.handle(&1, 1001, SaveThenDispatch),
154        Err(Error::Rejected)
155    );
156    assert_eq!(Total(&handler.store).handle(1), Ok(4000));
157    assert!(handler.publisher.accepted.is_empty());
158    handler.handle(&1, 200, SaveThenDispatch).unwrap();
159    assert_eq!(Total(&handler.store).handle(1), Ok(4200));
160    assert_eq!(handler.publisher.accepted, vec![Placed(1)]);
161    println!("cqrs: rejected invalid commands; query total=4200");
162}
cqrs/cqrs.rs

cqrs/
cqrs.rs
