mod event_payload;
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_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>,
error: Option<Path>,
},
Event(EventBinding),
}
fn classify_batpak_attr(attr: &Attribute) -> syn::Result<BatpakAttrKind> {
let mut input: Option<Path> = None;
let mut cache_version: Option<LitInt> = None;
let mut error_ty: Option<Path> = None;
let mut event: Option<Path> = None;
let mut handler: Option<Ident> = None;
attr.parse_nested_meta(|meta| {
let key = meta.path.get_ident().ok_or_else(|| {
meta.error("expected `input`, `cache_version`, `error`, `event`, or `handler`")
})?;
match key.to_string().as_str() {
"input" => {
if input.is_some() {
return Err(meta.error("duplicate `input` key within attribute"));
}
input = Some(meta.value()?.parse::<Path>()?);
}
"cache_version" => {
if cache_version.is_some() {
return Err(meta.error("duplicate `cache_version` key within attribute"));
}
cache_version = Some(meta.value()?.parse::<LitInt>()?);
}
"error" => {
if error_ty.is_some() {
return Err(meta.error("duplicate `error` key within attribute"));
}
error_ty = Some(meta.value()?.parse::<Path>()?);
}
"event" => {
if event.is_some() {
return Err(meta.error("duplicate `event` key within attribute"));
}
event = Some(meta.value()?.parse::<Path>()?);
}
"handler" => {
if handler.is_some() {
return Err(meta.error("duplicate `handler` key within attribute"));
}
handler = Some(meta.value()?.parse::<Ident>()?);
}
other => {
return Err(meta.error(format!(
"unknown key `{other}`, expected `input`, `cache_version`, `error`, `event`, or `handler`"
)));
}
}
Ok(())
})?;
let has_config = input.is_some() || cache_version.is_some() || error_ty.is_some();
let has_event = event.is_some() || 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`, `error`) or an event-binding pair (`event`, `handler`), not both",
));
}
if has_event {
let event = event.ok_or_else(|| {
syn::Error::new(
attr.span(),
"event-binding attribute is missing `event = <PayloadType>`",
)
})?;
let handler = 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`, `error`, or the `event`/`handler` pair",
));
}
Ok(BatpakAttrKind::Config {
input,
cache_version,
error: error_ty,
})
}
fn expand_event_sourced(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStream> {
match &input.data {
Data::Struct(s) => match &s.fields {
Fields::Named(_) => {}
Fields::Unnamed(f) => {
return Err(syn::Error::new(
f.span(),
"#[derive(EventSourced)] requires a named-field struct; tuple structs are not supported",
));
}
Fields::Unit => {
return Err(syn::Error::new(
input.ident.span(),
"#[derive(EventSourced)] requires a named-field struct; unit structs are not supported",
));
}
},
Data::Enum(e) => {
return Err(syn::Error::new(
e.enum_token.span,
"#[derive(EventSourced)] requires a named-field struct; enums are not supported",
));
}
Data::Union(u) => {
return Err(syn::Error::new(
u.union_token.span,
"#[derive(EventSourced)] requires a named-field struct; unions are not supported",
));
}
}
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 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,
error: attr_error,
} => {
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);
}
}
BatpakAttrKind::Event(binding) => {
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(),
"duplicate `event = X` — each payload type may be bound to exactly one handler per projection",
));
}
bindings.push(binding);
}
}
}
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",
));
}
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 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;
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> {
match &input.data {
Data::Struct(s) => match &s.fields {
Fields::Named(_) => {}
Fields::Unnamed(f) => {
return Err(syn::Error::new(
f.span(),
"#[derive(MultiEventReactor)] requires a named-field struct; tuple structs are not supported",
));
}
Fields::Unit => {
return Err(syn::Error::new(
input.ident.span(),
"#[derive(MultiEventReactor)] requires a named-field struct; unit structs are not supported",
));
}
},
Data::Enum(e) => {
return Err(syn::Error::new(
e.enum_token.span,
"#[derive(MultiEventReactor)] requires a named-field struct; enums are not supported",
));
}
Data::Union(u) => {
return Err(syn::Error::new(
u.union_token.span,
"#[derive(MultiEventReactor)] requires a named-field struct; unions are not supported",
));
}
}
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,
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(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) => {
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(),
"duplicate `event = X` — each payload type may be bound to exactly one handler per reactor",
));
}
bindings.push(binding);
}
}
}
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(())
}
}
})
}