mod event_payload;
mod operation;
use proc_macro::TokenStream;
use quote::quote;
use std::collections::HashSet;
use syn::{
parse_macro_input, spanned::Spanned, Attribute, Data, DeriveInput, Fields, Ident, LitInt, Path,
};
#[proc_macro_derive(EventPayload, attributes(batpak))]
pub fn derive_event_payload(input: TokenStream) -> TokenStream {
let input = parse_macro_input!(input as DeriveInput);
match event_payload::expand(&input) {
Ok(ts) => ts.into(),
Err(e) => e.to_compile_error().into(),
}
}
#[proc_macro_attribute]
pub fn operation(attr: TokenStream, item: TokenStream) -> TokenStream {
let args = parse_macro_input!(attr as operation::OperationArgs);
let function = parse_macro_input!(item as syn::ItemFn);
match operation::expand_operation(args, &function) {
Ok(tokens) => tokens.into(),
Err(error) => error.to_compile_error().into(),
}
}
#[proc_macro_derive(MultiEventReactor, attributes(batpak))]
pub fn derive_multi_event_reactor(input: TokenStream) -> TokenStream {
let input = parse_macro_input!(input as DeriveInput);
match expand_multi_event_reactor(&input) {
Ok(ts) => ts.into(),
Err(e) => e.to_compile_error().into(),
}
}
#[proc_macro_derive(EventSourced, attributes(batpak))]
pub fn derive_event_sourced(input: TokenStream) -> TokenStream {
let input = parse_macro_input!(input as DeriveInput);
match expand_event_sourced(&input) {
Ok(ts) => ts.into(),
Err(e) => e.to_compile_error().into(),
}
}
struct EventBinding {
event: Path,
handler: Ident,
}
enum BatpakAttrKind {
Config {
input: Option<Path>,
cache_version: Option<LitInt>,
state_max_cardinality: Option<LitInt>,
error: Option<Path>,
},
Event(EventBinding),
}
#[derive(Default)]
struct BatpakAttrParts {
input: Option<Path>,
cache_version: Option<LitInt>,
state_max_cardinality: Option<LitInt>,
error_ty: Option<Path>,
event: Option<Path>,
handler: Option<Ident>,
}
impl BatpakAttrParts {
fn set_nested(&mut self, meta: &syn::meta::ParseNestedMeta<'_>) -> syn::Result<()> {
let key = meta.path.get_ident().ok_or_else(|| {
meta.error("expected `input`, `cache_version`, `state_max_cardinality`, `error`, `event`, or `handler`")
})?;
let key_name = key.to_string();
if self.set_config_nested(key_name.as_str(), meta)? {
return Ok(());
}
if self.set_event_nested(key_name.as_str(), meta)? {
return Ok(());
}
Err(meta.error(format!(
"unknown key `{key_name}`, expected `input`, `cache_version`, `state_max_cardinality`, `error`, `event`, or `handler`"
)))
}
fn set_config_nested(
&mut self,
key: &str,
meta: &syn::meta::ParseNestedMeta<'_>,
) -> syn::Result<bool> {
match key {
"input" => {
if self.input.is_some() {
return Err(meta.error("duplicate `input` key within attribute"));
}
self.input = Some(meta.value()?.parse::<Path>()?);
}
"cache_version" => {
if self.cache_version.is_some() {
return Err(meta.error("duplicate `cache_version` key within attribute"));
}
self.cache_version = Some(meta.value()?.parse::<LitInt>()?);
}
"state_max_cardinality" => {
if self.state_max_cardinality.is_some() {
return Err(
meta.error("duplicate `state_max_cardinality` key within attribute")
);
}
self.state_max_cardinality = Some(meta.value()?.parse::<LitInt>()?);
}
"error" => {
if self.error_ty.is_some() {
return Err(meta.error("duplicate `error` key within attribute"));
}
self.error_ty = Some(meta.value()?.parse::<Path>()?);
}
_ => return Ok(false),
}
Ok(true)
}
fn set_event_nested(
&mut self,
key: &str,
meta: &syn::meta::ParseNestedMeta<'_>,
) -> syn::Result<bool> {
match key {
"event" => {
if self.event.is_some() {
return Err(meta.error("duplicate `event` key within attribute"));
}
self.event = Some(meta.value()?.parse::<Path>()?);
}
"handler" => {
if self.handler.is_some() {
return Err(meta.error("duplicate `handler` key within attribute"));
}
self.handler = Some(meta.value()?.parse::<Ident>()?);
}
_ => return Ok(false),
}
Ok(true)
}
fn finish(self, attr: &Attribute) -> syn::Result<BatpakAttrKind> {
let has_config = self.input.is_some()
|| self.cache_version.is_some()
|| self.state_max_cardinality.is_some()
|| self.error_ty.is_some();
let has_event = self.event.is_some() || self.handler.is_some();
if has_config && has_event {
return Err(syn::Error::new(
attr.span(),
"`#[batpak(...)]` attribute must contain either config keys \
(`input`, `cache_version`, `state_max_cardinality`, `error`) or an event-binding pair (`event`, `handler`), not both",
));
}
if has_event {
let event = self.event.ok_or_else(|| {
syn::Error::new(
attr.span(),
"event-binding attribute is missing `event = <PayloadType>`",
)
})?;
let handler = self.handler.ok_or_else(|| {
syn::Error::new(
attr.span(),
"event-binding attribute is missing `handler = <fn_name>`",
)
})?;
return Ok(BatpakAttrKind::Event(EventBinding { event, handler }));
}
if !has_config {
return Err(syn::Error::new(
attr.span(),
"`#[batpak(...)]` must contain at least one key: `input`, `cache_version`, `state_max_cardinality`, `error`, or the `event`/`handler` pair",
));
}
Ok(BatpakAttrKind::Config {
input: self.input,
cache_version: self.cache_version,
state_max_cardinality: self.state_max_cardinality,
error: self.error_ty,
})
}
}
fn classify_batpak_attr(attr: &Attribute) -> syn::Result<BatpakAttrKind> {
let mut parts = BatpakAttrParts::default();
attr.parse_nested_meta(|meta| parts.set_nested(&meta))?;
parts.finish(attr)
}
fn ensure_named_field_struct(input: &DeriveInput, derive_name: &str) -> syn::Result<()> {
match &input.data {
Data::Struct(s) => match &s.fields {
Fields::Named(_) => Ok(()),
Fields::Unnamed(f) => Err(syn::Error::new(
f.span(),
format!(
"#[derive({derive_name})] requires a named-field struct; tuple structs are not supported"
),
)),
Fields::Unit => Err(syn::Error::new(
input.ident.span(),
format!(
"#[derive({derive_name})] requires a named-field struct; unit structs are not supported"
),
)),
},
Data::Enum(e) => Err(syn::Error::new(
e.enum_token.span,
format!("#[derive({derive_name})] requires a named-field struct; enums are not supported"),
)),
Data::Union(u) => Err(syn::Error::new(
u.union_token.span,
format!(
"#[derive({derive_name})] requires a named-field struct; unions are not supported"
),
)),
}
}
struct EventSourcedDeriveAttrs {
input_path: Path,
cache_version_lit: Option<LitInt>,
state_max_cardinality_lit: Option<LitInt>,
bindings: Vec<EventBinding>,
}
fn collect_event_sourced_attrs(input: &DeriveInput) -> syn::Result<EventSourcedDeriveAttrs> {
let batpak_attrs: Vec<&Attribute> = input
.attrs
.iter()
.filter(|a| a.path().is_ident("batpak"))
.collect();
if batpak_attrs.is_empty() {
return Err(syn::Error::new(
input.ident.span(),
"#[derive(EventSourced)] requires at least one `#[batpak(input = <Lane>)]` attribute",
));
}
let mut input_path: Option<Path> = None;
let mut cache_version_lit: Option<LitInt> = None;
let mut state_max_cardinality_lit: Option<LitInt> = None;
let mut bindings: Vec<EventBinding> = Vec::new();
let mut seen_events: HashSet<String> = HashSet::new();
for attr in &batpak_attrs {
match classify_batpak_attr(attr)? {
BatpakAttrKind::Config {
input: attr_input,
cache_version: attr_cache,
state_max_cardinality: attr_state_max,
error: attr_error,
} => {
collect_event_sourced_config(
&mut input_path,
&mut cache_version_lit,
&mut state_max_cardinality_lit,
attr_input,
attr_cache,
attr_state_max,
attr_error,
)?;
}
BatpakAttrKind::Event(binding) => {
collect_unique_event_binding(
&mut bindings,
&mut seen_events,
binding,
"projection",
)?;
}
}
}
let input_path = input_path.ok_or_else(|| {
syn::Error::new(
input.ident.span(),
"#[derive(EventSourced)] requires `#[batpak(input = <Lane>)]` — e.g. `input = JsonValueInput` or `input = RawMsgpackInput`",
)
})?;
if bindings.is_empty() {
return Err(syn::Error::new(
input.ident.span(),
"`#[derive(EventSourced)]` requires at least one `#[batpak(event = T, handler = h)]` binding",
));
}
Ok(EventSourcedDeriveAttrs {
input_path,
cache_version_lit,
state_max_cardinality_lit,
bindings,
})
}
fn collect_event_sourced_config(
input_path: &mut Option<Path>,
cache_version_lit: &mut Option<LitInt>,
state_max_cardinality_lit: &mut Option<LitInt>,
attr_input: Option<Path>,
attr_cache: Option<LitInt>,
attr_state_max: Option<LitInt>,
attr_error: Option<Path>,
) -> syn::Result<()> {
if let Some(path) = attr_error {
return Err(syn::Error::new(
path.span(),
"`error` is not valid on `#[derive(EventSourced)]` — projections do not have an associated error type",
));
}
if let Some(path) = attr_input {
if input_path.is_some() {
return Err(syn::Error::new(
path.span(),
"duplicate `input =` across `#[batpak(...)]` config attributes — `input` must appear exactly once",
));
}
*input_path = Some(path);
}
if let Some(lit) = attr_cache {
if cache_version_lit.is_some() {
return Err(syn::Error::new(
lit.span(),
"duplicate `cache_version =` across `#[batpak(...)]` config attributes",
));
}
*cache_version_lit = Some(lit);
}
if let Some(lit) = attr_state_max {
if state_max_cardinality_lit.is_some() {
return Err(syn::Error::new(
lit.span(),
"duplicate `state_max_cardinality =` across `#[batpak(...)]` config attributes",
));
}
*state_max_cardinality_lit = Some(lit);
}
Ok(())
}
fn collect_unique_event_binding(
bindings: &mut Vec<EventBinding>,
seen_events: &mut HashSet<String>,
binding: EventBinding,
owner: &str,
) -> syn::Result<()> {
require_single_segment_event_path(&binding.event)?;
let key = binding.event.to_token_stream_string();
if !seen_events.insert(key) {
return Err(syn::Error::new(
binding.event.span(),
format!(
"duplicate `event = X` — each payload type may be bound to exactly one handler per {owner}"
),
));
}
bindings.push(binding);
Ok(())
}
fn expand_event_sourced(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStream> {
ensure_named_field_struct(input, "EventSourced")?;
let attrs = collect_event_sourced_attrs(input)?;
let input_path = attrs.input_path;
let cache_version_lit = attrs.cache_version_lit;
let state_max_cardinality_lit = attrs.state_max_cardinality_lit;
let bindings = attrs.bindings;
let cache_version_value: u64 = match &cache_version_lit {
Some(lit) => lit.base10_parse::<u64>()?,
None => 0u64,
};
let ident = &input.ident;
let (impl_generics, ty_generics, where_clause) = input.generics.split_for_impl();
let state_contract_impl = match &state_max_cardinality_lit {
Some(lit) => {
let state_max_cardinality_value = lit.base10_parse::<u64>()?;
if state_max_cardinality_value != 1 {
return Err(syn::Error::new_spanned(
lit,
"#[derive(EventSourced)] supports only single-aggregate state (n = 1); \
implement EventSourced by hand with a real `state_extent()` for multi-key state",
));
}
quote! {
const STATE_CONTRACT: ::batpak::event::ProjectionStateContract =
::batpak::event::ProjectionStateContract::Bounded {
key_space: ::core::concat!(
::core::module_path!(),
"::",
::core::stringify!(#ident)
),
max_cardinality: #state_max_cardinality_value,
retention_policy: "derive-event-sourced-state-object",
compaction_policy: "projection-cache-overwrite",
checkpoint_policy: "projection-cache",
};
fn state_extent(&self) -> ::batpak::event::StateExtent {
let _ = self;
::batpak::event::StateExtent::cardinality(
1,
::batpak::event::StateExtentCost::ConstantTime,
)
}
}
}
None => quote! {},
};
let arms: Vec<proc_macro2::TokenStream> = bindings
.iter()
.map(|b| {
let event_ty = &b.event;
let handler_fn = &b.handler;
quote! {
match ::batpak::event::DecodeTyped::route_typed::<#event_ty>(event) {
::core::result::Result::Ok(::core::option::Option::Some(__p)) => {
self.#handler_fn(&__p);
return;
}
::core::result::Result::Ok(::core::option::Option::None) => {}
::core::result::Result::Err(__e) => {
::core::panic!(
"EventSourced: decode failed for matched kind {}: {}",
::core::stringify!(#event_ty),
__e
);
}
}
}
})
.collect();
let kind_exprs: Vec<proc_macro2::TokenStream> = bindings
.iter()
.map(|b| {
let event_ty = &b.event;
quote! {
<#event_ty as ::batpak::event::EventPayload>::KIND
}
})
.collect();
let kind_count = bindings.len();
let handler_checks: Vec<proc_macro2::TokenStream> = bindings
.iter()
.map(|b| {
let event_ty = &b.event;
let handler_fn = &b.handler;
quote! {
let _: fn(&mut Self, &#event_ty) = Self::#handler_fn;
}
})
.collect();
let input_assertion = {
quote! {
const _: fn() = || {
fn __batpak_assert_projection_input<T: ::batpak::event::ProjectionInput>() {}
__batpak_assert_projection_input::<#input_path>();
};
}
};
Ok(quote! {
#input_assertion
impl #impl_generics ::batpak::event::EventSourced for #ident #ty_generics #where_clause {
type Input = #input_path;
#state_contract_impl
fn from_events(
events: &[::batpak::event::ProjectionEvent<Self>],
) -> ::core::option::Option<Self> {
if events.is_empty() {
return ::core::option::Option::None;
}
let mut state: Self = ::core::default::Default::default();
for __ev in events {
state.apply_event(__ev);
}
::core::option::Option::Some(state)
}
fn apply_event(&mut self, event: &::batpak::event::ProjectionEvent<Self>) {
#(#handler_checks)*
#(#arms)*
let _ = event;
}
fn relevant_event_kinds() -> &'static [::batpak::event::EventKind] {
static KINDS: [::batpak::event::EventKind; #kind_count] = [
#(#kind_exprs),*
];
&KINDS
}
fn schema_version() -> u64 {
#cache_version_value
}
}
})
}
trait ToTokenStreamString {
fn to_token_stream_string(&self) -> String;
}
impl ToTokenStreamString for Path {
fn to_token_stream_string(&self) -> String {
quote!(#self).to_string()
}
}
fn require_single_segment_event_path(path: &Path) -> syn::Result<()> {
if path.leading_colon.is_some() || path.segments.len() != 1 {
return Err(syn::Error::new_spanned(
path,
"event type must be named by its in-scope single-segment name — use a `use` import if the type is in another module",
));
}
Ok(())
}
fn expand_multi_event_reactor(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStream> {
ensure_named_field_struct(input, "MultiEventReactor")?;
let batpak_attrs: Vec<&Attribute> = input
.attrs
.iter()
.filter(|a| a.path().is_ident("batpak"))
.collect();
if batpak_attrs.is_empty() {
return Err(syn::Error::new(
input.ident.span(),
"#[derive(MultiEventReactor)] requires `#[batpak(input = <Lane>)]` plus at least one `#[batpak(event = <Payload>, handler = <fn>)]` attribute",
));
}
let mut input_path: Option<Path> = None;
let mut error_path: Option<Path> = None;
let mut bindings: Vec<EventBinding> = Vec::new();
let mut seen_events: HashSet<String> = HashSet::new();
for attr in &batpak_attrs {
match classify_batpak_attr(attr)? {
BatpakAttrKind::Config {
input: attr_input,
cache_version,
state_max_cardinality,
error: attr_error,
} => {
if let Some(lit) = cache_version {
return Err(syn::Error::new(
lit.span(),
"`cache_version` is not valid on `#[derive(MultiEventReactor)]` — \
`cache_version` is a projection-cache key, not a reactor setting",
));
}
if let Some(lit) = state_max_cardinality {
return Err(syn::Error::new(
lit.span(),
"`state_max_cardinality` is not valid on `#[derive(MultiEventReactor)]` — \
state cardinality is a projection contract, not a reactor setting",
));
}
if let Some(path) = attr_input {
if input_path.is_some() {
return Err(syn::Error::new(
path.span(),
"duplicate `input =` across `#[batpak(...)]` config attributes — `input` must appear exactly once",
));
}
input_path = Some(path);
}
if let Some(path) = attr_error {
if error_path.is_some() {
return Err(syn::Error::new(
path.span(),
"duplicate `error =` across `#[batpak(...)]` config attributes — `error` must appear exactly once",
));
}
error_path = Some(path);
}
}
BatpakAttrKind::Event(binding) => {
collect_unique_event_binding(&mut bindings, &mut seen_events, binding, "reactor")?;
}
}
}
let input_path = input_path.ok_or_else(|| {
syn::Error::new(
input.ident.span(),
"#[derive(MultiEventReactor)] requires `#[batpak(input = <Lane>)]` — e.g. `input = JsonValueInput` or `input = RawMsgpackInput`",
)
})?;
let error_path = error_path.ok_or_else(|| {
syn::Error::new(
input.ident.span(),
"#[derive(MultiEventReactor)] requires `#[batpak(error = <ErrorType>)]` — the shared error type all handlers return",
)
})?;
if bindings.is_empty() {
return Err(syn::Error::new(
input.ident.span(),
"#[derive(MultiEventReactor)] requires at least one `#[batpak(event = <Payload>, handler = <fn>)]`",
));
}
let ident = &input.ident;
let (impl_generics, ty_generics, where_clause) = input.generics.split_for_impl();
let kind_exprs: Vec<proc_macro2::TokenStream> = bindings
.iter()
.map(|b| {
let event_ty = &b.event;
quote! {
<#event_ty as ::batpak::event::EventPayload>::KIND
}
})
.collect();
let kind_count = bindings.len();
let arms: Vec<proc_macro2::TokenStream> = bindings
.iter()
.map(|b| {
let event_ty = &b.event;
let handler_fn = &b.handler;
quote! {
match ::batpak::event::DecodeTyped::route_typed::<#event_ty>(&event.event) {
::core::result::Result::Ok(::core::option::Option::Some(__p)) => {
let __typed_event = ::batpak::event::StoredEvent {
coordinate: event.coordinate.clone(),
event: ::batpak::event::Event {
header: event.event.header.clone(),
payload: __p,
hash_chain: event.event.hash_chain.clone(),
},
};
return self
.#handler_fn(&__typed_event, out, at_least_once)
.map_err(::batpak::event::MultiDispatchError::User);
}
::core::result::Result::Ok(::core::option::Option::None) => {}
::core::result::Result::Err(__e) => {
return ::core::result::Result::Err(
::batpak::event::MultiDispatchError::Decode(__e)
);
}
}
}
})
.collect();
let handler_checks: Vec<proc_macro2::TokenStream> = bindings
.iter()
.map(|b| {
let event_ty = &b.event;
let handler_fn = &b.handler;
quote! {
let _: fn(
&mut Self,
&::batpak::event::StoredEvent<#event_ty>,
&mut ::batpak::store::ReactionBatch,
::core::option::Option<&::batpak::store::AtLeastOnce>,
) -> ::core::result::Result<(), #error_path> = Self::#handler_fn;
}
})
.collect();
let attr_assertions = {
quote! {
const _: fn() = || {
fn __batpak_assert_projection_input<T: ::batpak::event::ProjectionInput>() {}
__batpak_assert_projection_input::<#input_path>();
};
const _: fn() = || {
fn __batpak_assert_error<
T: ::core::marker::Send
+ ::core::marker::Sync
+ 'static
+ ::std::error::Error,
>() {}
__batpak_assert_error::<#error_path>();
};
}
};
Ok(quote! {
#attr_assertions
impl #impl_generics ::batpak::event::MultiReactive<#input_path>
for #ident #ty_generics #where_clause
{
type Error = #error_path;
fn relevant_event_kinds() -> &'static [::batpak::event::EventKind] {
static KINDS: [::batpak::event::EventKind; #kind_count] = [
#(#kind_exprs),*
];
&KINDS
}
fn dispatch(
&mut self,
event: &::batpak::event::StoredEvent<
<#input_path as ::batpak::event::ProjectionInput>::Payload,
>,
out: &mut ::batpak::store::ReactionBatch,
at_least_once: ::core::option::Option<&::batpak::store::AtLeastOnce>,
) -> ::core::result::Result<(), ::batpak::event::MultiDispatchError<Self::Error>> {
#(#handler_checks)*
#(#arms)*
::core::result::Result::Ok(())
}
}
})
}