イベントの定義

use std::hash::DefaultHasher;
use wakuwaku::iggy::{IggySuperPartitionName, PartitionKey};

/// 注文をキーとするイベントはすべて `orders` トピックに送られる。
#[derive(Hash)]
pub struct OrderId(pub u64);

impl PartitionKey for OrderId {
    type PartitionMerge = DefaultHasher;
    const SUPER_PARTITION_NAME: IggySuperPartitionName = IggySuperPartitionName("orders");
}

use std::hash::Hasher;
use wakuwaku::iggy::{IggySuperPartitionName, PartitionKey};

/// 64 ビット FNV-1a。アルゴリズムが固定・文書化されているので、
/// どの Rust バージョンでビルドしたパブリッシャーでも結果が一致する。
pub struct Fnv1a(u64);

impl Default for Fnv1a {
    fn default() -> Self {
        Self(0xcbf2_9ce4_8422_2325)
    }
}

impl Hasher for Fnv1a {
    fn write(&mut self, bytes: &[u8]) {
        for &byte in bytes {
            self.0 ^= u64::from(byte);
            self.0 = self.0.wrapping_mul(0x0000_0100_0000_01b3);
        }
    }

    fn finish(&self) -> u64 {
        self.0
    }
}

#[derive(Hash)]
pub struct CustomerId(pub u64);

impl PartitionKey for CustomerId {
    type PartitionMerge = Fnv1a;
    const SUPER_PARTITION_NAME: IggySuperPartitionName = IggySuperPartitionName("customers");
}

use kanau::{JsonMessageDe, JsonMessageSer};
use serde::{Deserialize, Serialize};
use wakuwaku::iggy::{Event, EventAtomicOrdering, EventBody, EventTypeTag, NonAssociative};

/// 注文が作成された。その注文の他のどのイベントよりも先に処理する必要がある。
#[derive(Debug, Clone, Serialize, Deserialize, JsonMessageSer, JsonMessageDe)]
pub struct OrderPlaced {
    pub order_id: u64,
    pub customer: String,
    pub total_cents: i64,
}

impl EventBody for OrderPlaced {}

impl Event for OrderPlaced {
    const TYPE_TAG: EventTypeTag = EventTypeTag::new(1);
    type Key = OrderId;
    type Algebra = NonAssociative;

    fn key(&self) -> OrderId {
        OrderId(self.order_id)
    }

    fn atomic_ordering(&self) -> EventAtomicOrdering {
        EventAtomicOrdering::Acquire
    }
}

use kanau::{JsonMessageDe, JsonMessageSer};
use serde::{Deserialize, Serialize};
use wakuwaku::iggy::{
    Associative, Event, EventAtomicOrdering, EventBody, EventSemigroup, EventTypeTag,
};

/// 支払いを受け取った。1 つの注文の支払いは合算できる。
#[derive(Debug, Clone, Serialize, Deserialize, JsonMessageSer, JsonMessageDe)]
pub struct PaymentReceived {
    pub order_id: u64,
    pub amount_cents: i64,
}

impl EventBody for PaymentReceived {}

impl EventSemigroup for PaymentReceived {
    fn combine(self, other: Self) -> Self {
        Self {
            order_id: self.order_id,
            amount_cents: self.amount_cents + other.amount_cents,
        }
    }
}

impl Event for PaymentReceived {
    const TYPE_TAG: EventTypeTag = EventTypeTag::new(2);
    type Key = OrderId;
    type Algebra = Associative;

    fn key(&self) -> OrderId {
        OrderId(self.order_id)
    }

    fn atomic_ordering(&self) -> EventAtomicOrdering {
        EventAtomicOrdering::Relaxed
    }
}

pub trait EventBody: kanau::message::MessageDe + kanau::message::MessageSer {}

[dependencies]
wakuwaku = { version = "1.0.0-rc3", features = ["iggy"] }
kanau = { version = "0.6", features = ["serde_json"] }
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["full"] }

use kanau::{JsonMessageDe, JsonMessageSer};
use serde::{Deserialize, Serialize};
use wakuwaku::iggy::EventBody;

#[derive(Debug, Clone, Serialize, Deserialize, JsonMessageSer, JsonMessageDe)]
pub struct OrderPlaced {
    pub order_id: u64,
    pub customer: String,
    pub total_cents: i64,
}

impl EventBody for OrderPlaced {}

use wakuwaku::deps::kanau::message::{DeserializeError, MessageDe, MessageSer, SerializeError};
use wakuwaku::iggy::EventBody;

/// 手書きのボディ。顧客 ID の 8 バイト（リトルエンディアン）。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct CustomerTouched {
    pub customer: u64,
}

impl MessageSer for CustomerTouched {
    type SerError = SerializeError;

    fn to_bytes(self) -> Result<Box<[u8]>, SerializeError> {
        Ok(Box::new(self.customer.to_le_bytes()))
    }
}

impl MessageDe for CustomerTouched {
    type DeError = DeserializeError;

    fn from_bytes(bytes: &[u8]) -> Result<Self, DeserializeError> {
        let bytes: [u8; 8] = bytes
            .try_into()
            .map_err(|error: std::array::TryFromSliceError| DeserializeError(error.into()))?;
        Ok(Self {
            customer: u64::from_le_bytes(bytes),
        })
    }
}

impl EventBody for CustomerTouched {}

use wakuwaku::iggy::{
    Event, EventAtomicOrdering, EventTypeTag, IdempotentCommutative, IntoEventBody,
};

/// 別のワイヤ用ボディとして発行されるドメイン型。
pub struct Customer {
    pub id: u64,
    pub name: String,
    pub password_hash: String,
}

/// イベント「この顧客が変更された」。ワイヤに載るのは ID だけ。
pub struct CustomerChanged(pub Customer);

impl IntoEventBody for CustomerChanged {
    type Target = CustomerTouched;

    fn into_event_body(self) -> CustomerTouched {
        CustomerTouched { customer: self.0.id }
    }
}

impl Event for CustomerChanged {
    const TYPE_TAG: EventTypeTag = EventTypeTag::new(1);
    type Key = CustomerId;
    type Algebra = IdempotentCommutative;

    fn key(&self) -> CustomerId {
        CustomerId(self.0.id)
    }

    fn atomic_ordering(&self) -> EventAtomicOrdering {
        EventAtomicOrdering::Relaxed
    }
}

/// `CustomerChanged` のハンドラはデコード済みのボディ型を受け取る。
pub struct Reindex;

impl wakuwaku::iggy::EventHandler<CustomerChanged> for Reindex {
    const MAX_BATCH: std::num::NonZeroUsize = std::num::NonZeroUsize::new(50).unwrap();

    async fn handle(
        &self,
        run: &[CustomerTouched],
    ) -> Result<(), wakuwaku::iggy::error::HandleError> {
        for touched in run {
            println!("reindex customer {}", touched.customer);
        }
        Ok(())
    }
}