Move connection events to reducers (#309)

* Move connection events to reducers

* More rebase fixups

* Avoid double-reference

* Filter out special reducers in generate cli

This updates filtering of `__init__` to exclude all special reducers, as well as moves the filtering to centralised place before calling language-specific generate command.

* Incrememted ABI version number

---------

Co-authored-by: Tyler Cloutier <cloutiertyler@aol.com>

crates/bindings-csharp/Codegen/Codegen.csproj

@@ -7,7 +7,7 @@
     <LangVersion>latest</LangVersion>
     <Nullable>enable</Nullable>
     <AssemblyName>SpacetimeDB.Codegen</AssemblyName>
-    <AssemblyVersion>0.7.0</AssemblyVersion>
+    <AssemblyVersion>0.7.1</AssemblyVersion>
   </PropertyGroup>
 
   <PropertyGroup>

crates/bindings-csharp/Runtime/Module.cs

@@ -142,6 +142,8 @@ public static class ReducerKind
 {
     public const string Init = "__init__";
     public const string Update = "__update__";
+    public const string Connect = "__identity_connected__";
+    public const string Disconnect = "__identity_disconnected__";
 }
 
 public interface IReducer

crates/bindings-csharp/Runtime/Runtime.cs

@@ -276,37 +276,6 @@ public DbEventArgs(byte[] senderIdentity, byte[] senderAddress, ulong timestamp_
         }
     }
 
-    public static event Action<DbEventArgs>? OnConnect;
-    public static event Action<DbEventArgs>? OnDisconnect;
-
-    // Note: this is accessed by C bindings.
-    private static string? IdentityConnected(byte[] sender_identity, byte[] sender_address, ulong timestamp)
-    {
-        try
-        {
-            OnConnect?.Invoke(new(sender_identity, sender_address, timestamp));
-            return null;
-        }
-        catch (Exception e)
-        {
-            return e.ToString();
-        }
-    }
-
-    // Note: this is accessed by C bindings.
-    private static string? IdentityDisconnected(byte[] sender_identity, byte[] sender_address, ulong timestamp)
-    {
-        try
-        {
-            OnDisconnect?.Invoke(new(sender_identity, sender_address, timestamp));
-            return null;
-        }
-        catch (Exception e)
-        {
-            return e.ToString();
-        }
-    }
-
     [MethodImpl(MethodImplOptions.InternalCall)]
     private extern static void ScheduleReducer(
         string name,

crates/bindings-csharp/Runtime/Runtime.csproj

@@ -5,7 +5,7 @@
     <LangVersion>latest</LangVersion>
     <Nullable>enable</Nullable>
     <AssemblyName>SpacetimeDB.Runtime</AssemblyName>
-    <AssemblyVersion>0.7.0</AssemblyVersion>
+    <AssemblyVersion>0.7.1</AssemblyVersion>
   </PropertyGroup>
 
   <PropertyGroup>

crates/bindings-csharp/Runtime/bindings.c

@@ -779,5 +779,5 @@ __preinit__10_init_csharp() {
 
 // __attribute__((export_name("SPACETIME_ABI_VERSION"))) -
 // doesn't work on non-functions, must specify on command line
-const uint32_t SPACETIME_ABI_VERSION = /* 4.0 */ (4 << 16) | 0;
+const uint32_t SPACETIME_ABI_VERSION = /* 5.0 */ (5 << 16) | 0;
 const uint8_t SPACETIME_ABI_VERSION_IS_ADDR = 1;

crates/bindings-macro/src/lib.rs

@@ -103,11 +103,11 @@ fn route_input(input: MacroInput, item: TokenStream) -> syn::Result<TokenStream>
         MacroInput::Table => spacetimedb_table(item),
         MacroInput::Init => spacetimedb_init(item),
         MacroInput::Reducer { repeat } => spacetimedb_reducer(repeat, item),
-        MacroInput::Connect => spacetimedb_connect_disconnect(item, true),
-        MacroInput::Disconnect => spacetimedb_connect_disconnect(item, false),
-        MacroInput::Migrate => spacetimedb_migrate(item),
+        MacroInput::Connect => spacetimedb_special_reducer("__identity_connected__", item),
+        MacroInput::Disconnect => spacetimedb_special_reducer("__identity_disconnected__", item),
+        MacroInput::Migrate => spacetimedb_special_reducer("__migrate__", item),
         MacroInput::Index { ty, name, field_names } => spacetimedb_index(ty, name, field_names, item),
-        MacroInput::Update => spacetimedb_update(item),
+        MacroInput::Update => spacetimedb_special_reducer("__update__", item),
     }
 }
 
@@ -264,16 +264,12 @@ mod kw {
 
 /// Generates a reducer in place of `item`.
 fn spacetimedb_reducer(repeat: Option<Duration>, item: TokenStream) -> syn::Result<TokenStream> {
-    // TODO(kim): Find a better place for these. `core/host/wasm_common.rs` has similar
-    // definitions, but we can't depend on `core` here.
-    const RESERVED_REDUCER_NAMES: &[&str] = &["__init__", "__migrate__", "__update__"];
-
-    let repeat_dur = repeat.map_or(ReducerExtra::None, ReducerExtra::Repeat);
+    let repeat_dur = repeat.map_or(ReducerExtra::Schedule, ReducerExtra::Repeat);
     let original_function = syn::parse2::<ItemFn>(item)?;
 
-    // Extract reducer name, making sure it's not `__init__`.
+    // Extract reducer name, making sure it's not `__XXX__` as that's the form we reserve for special reducers.
     let reducer_name = original_function.sig.ident.to_string();
-    if RESERVED_REDUCER_NAMES.iter().any(|name| name == &reducer_name) {
+    if reducer_name.starts_with("__") && reducer_name.ends_with("__") {
         return Err(syn::Error::new_spanned(
             &original_function.sig.ident,
             "reserved reducer name",
@@ -287,13 +283,13 @@ fn spacetimedb_reducer(repeat: Option<Duration>, item: TokenStream) -> syn::Resu
 fn spacetimedb_init(item: TokenStream) -> syn::Result<TokenStream> {
     let original_function = syn::parse2::<ItemFn>(item)?;
 
-    gen_reducer(original_function, "__init__", ReducerExtra::Init)
+    gen_reducer(original_function, "__init__", ReducerExtra::None)
 }
 
 enum ReducerExtra {
     None,
+    Schedule,
     Repeat(Duration),
-    Init,
 }
 
 fn gen_reducer(original_function: ItemFn, reducer_name: &str, extra: ReducerExtra) -> syn::Result<TokenStream> {
@@ -321,14 +317,18 @@ fn gen_reducer(original_function: ItemFn, reducer_name: &str, extra: ReducerExtr
     // }
 
     // Extract all function parameters, except for `self` ones that aren't allowed.
-    let args = original_function.sig.inputs.iter().map(|x| match x {
-        // TODO: improve error message!
-        FnArg::Receiver(_) => panic!(),
-        FnArg::Typed(x) => x,
-    });
+    let typed_args = original_function
+        .sig
+        .inputs
+        .iter()
+        .map(|arg| match arg {
+            FnArg::Typed(arg) => Ok(arg),
+            _ => Err(syn::Error::new_spanned(arg, "expected typed argument")),
+        })
+        .collect::<syn::Result<Vec<_>>>()?;
 
     // Extract all function parameter names.
-    let arg_names = args.clone().map(|arg| {
+    let opt_arg_names = typed_args.iter().map(|arg| {
         if let syn::Pat::Ident(i) = &*arg.pat {
             let name = i.ident.to_string();
             quote!(Some(#name))
@@ -337,8 +337,7 @@ fn gen_reducer(original_function: ItemFn, reducer_name: &str, extra: ReducerExtr
         }
     });
 
-    // Extract all function parameter types.
-    let arg_tys = args.map(|arg| &arg.ty);
+    let arg_tys = typed_args.iter().map(|arg| arg.ty.as_ref()).collect::<Vec<_>>();
 
     // Extract the return type.
     let ret_ty = match &original_function.sig.output {
@@ -349,34 +348,49 @@ fn gen_reducer(original_function: ItemFn, reducer_name: &str, extra: ReducerExtr
 
     let register_describer_symbol = format!("__preinit__20_register_describer_{reducer_name}");
 
-    let (epilogue, repeater_impl) = match &extra {
-        ReducerExtra::None | ReducerExtra::Init => (quote!(), quote!()),
-        ReducerExtra::Repeat(repeat_dur) => {
-            let repeat_dur = duration_totokens(*repeat_dur);
-            let epilogue = quote! {
-                if _res.is_ok() {
-                    spacetimedb::rt::schedule_repeater::<_, _, #func_name>(#func_name)
-                }
-            };
-            let repeater_impl = quote! {
-                impl spacetimedb::rt::RepeaterInfo for #func_name {
-                    const REPEAT_INTERVAL: ::core::time::Duration = #repeat_dur;
-                }
-            };
-            (epilogue, repeater_impl)
-        }
-    };
+    let mut epilogue = TokenStream::new();
+    let mut extra_impls = TokenStream::new();
+
+    if !matches!(extra, ReducerExtra::None) {
+        let arg_names = typed_args
+            .iter()
+            .enumerate()
+            .map(|(i, arg)| match &*arg.pat {
+                syn::Pat::Ident(pat) => pat.ident.clone(),
+                _ => format_ident!("__arg{}", i),
+            })
+            .collect::<Vec<_>>();
+
+        extra_impls.extend(quote!(impl #func_name {
+            pub fn schedule(__time: spacetimedb::Timestamp #(, #arg_names: #arg_tys)*) -> spacetimedb::ScheduleToken<#func_name> {
+                spacetimedb::rt::schedule(__time, (#(#arg_names,)*))
+            }
+        }));
+    }
+
+    if let ReducerExtra::Repeat(repeat_dur) = &extra {
+        let repeat_dur = duration_totokens(*repeat_dur);
+        epilogue.extend(quote! {
+            if _res.is_ok() {
+                spacetimedb::rt::schedule_repeater::<_, _, #func_name>(#func_name)
+            }
+        });
+        extra_impls.extend(quote! {
+            impl spacetimedb::rt::RepeaterInfo for #func_name {
+                const REPEAT_INTERVAL: ::core::time::Duration = #repeat_dur;
+            }
+        });
+    }
 
     let generated_function = quote! {
-        // NOTE: double-underscoring names here is unnecessary, as Rust macros are hygienic.
         fn __reducer(
             __sender: spacetimedb::sys::Buffer,
             __caller_address: spacetimedb::sys::Buffer,
             __timestamp: u64,
             __args: &[u8]
         ) -> spacetimedb::sys::Buffer {
-            #(spacetimedb::rt::assert_reducerarg::<#arg_tys>();)*
-            #(spacetimedb::rt::assert_reducerret::<#ret_ty>();)*
+            #(spacetimedb::rt::assert_reducer_arg::<#arg_tys>();)*
+            #(spacetimedb::rt::assert_reducer_ret::<#ret_ty>();)*
             spacetimedb::rt::invoke_reducer(
                 #func_name,
                 __sender,
@@ -395,56 +409,21 @@ fn gen_reducer(original_function: ItemFn, reducer_name: &str, extra: ReducerExtr
         }
     };
 
-    let mut schedule_func_sig = original_function.sig.clone();
-    let schedule_func_body = {
-        schedule_func_sig.ident = format_ident!("schedule");
-        schedule_func_sig.output = syn::ReturnType::Type(
-            Token![->](Span::call_site()),
-            Box::new(syn::parse_quote!(spacetimedb::ScheduleToken<#func_name>)),
-        );
-        let arg_names = schedule_func_sig.inputs.iter_mut().enumerate().map(|(i, arg)| {
-            let syn::FnArg::Typed(arg) = arg else { panic!() };
-            match &mut *arg.pat {
-                syn::Pat::Ident(id) => {
-                    id.by_ref = None;
-                    id.mutability = None;
-                    id.ident.clone()
-                }
-                _ => {
-                    let ident = format_ident!("__arg{}", i);
-                    arg.pat = Box::new(syn::parse_quote!(#ident));
-                    ident
-                }
-            }
-        });
-        let schedule_args = quote!((#(#arg_names,)*));
-        let time_arg = format_ident!("__time");
-        schedule_func_sig
-            .inputs
-            .insert(0, syn::parse_quote!(#time_arg: spacetimedb::Timestamp));
-        quote! {
-            spacetimedb::rt::schedule(#time_arg, #schedule_args)
-        }
-    };
-
     Ok(quote! {
         const _: () = {
             #generated_describe_function
         };
         #[allow(non_camel_case_types)]
         #vis struct #func_name { _never: ::core::convert::Infallible }
-        impl #func_name {
-            #vis #schedule_func_sig { #schedule_func_body }
-        }
         impl spacetimedb::rt::ReducerInfo for #func_name {
             const NAME: &'static str = #reducer_name;
-            const ARG_NAMES: &'static [Option<&'static str>] = &[#(#arg_names),*];
+            const ARG_NAMES: &'static [Option<&'static str>] = &[#(#opt_arg_names),*];
             const INVOKE: spacetimedb::rt::ReducerFn = {
                 #generated_function
                 __reducer
             };
         }
-        #repeater_impl
+        #extra_impls
         #original_function
     })
 }
@@ -880,48 +859,9 @@ fn spacetimedb_index(
     Ok(output)
 }
 
-fn spacetimedb_migrate(item: TokenStream) -> syn::Result<TokenStream> {
-    let original_function = syn::parse2::<ItemFn>(item)?;
-    gen_reducer(original_function, "__migrate__", ReducerExtra::None)
-}
-
-fn spacetimedb_update(item: TokenStream) -> syn::Result<TokenStream> {
+fn spacetimedb_special_reducer(name: &str, item: TokenStream) -> syn::Result<TokenStream> {
     let original_function = syn::parse2::<ItemFn>(item)?;
-    gen_reducer(original_function, "__update__", ReducerExtra::None)
-}
-
-fn spacetimedb_connect_disconnect(item: TokenStream, connect: bool) -> syn::Result<TokenStream> {
-    let original_function = syn::parse2::<ItemFn>(item)?;
-    let func_name = &original_function.sig.ident;
-    let connect_disconnect_symbol = if connect {
-        "__identity_connected__"
-    } else {
-        "__identity_disconnected__"
-    };
-
-    let emission = quote! {
-        const _: () = {
-            #[export_name = #connect_disconnect_symbol]
-            extern "C" fn __connect_disconnect(
-                __sender: spacetimedb::sys::Buffer,
-                __caller_address: spacetimedb::sys::Buffer,
-                __timestamp: u64,
-            ) -> spacetimedb::sys::Buffer {
-                spacetimedb::rt::invoke_connection_func(#func_name, __sender, __caller_address, __timestamp)
-            }
-        };
-
-        #original_function
-    };
-
-    if std::env::var("PROC_MACRO_DEBUG").is_ok() {
-        {
-            #![allow(clippy::disallowed_macros)]
-            println!("{}", emission);
-        }
-    }
-
-    Ok(emission)
+    gen_reducer(original_function, name, ReducerExtra::None)
 }
 
 #[proc_macro]

crates/bindings-sys/src/lib.rs

@@ -34,7 +34,7 @@ use alloc::boxed::Box;
 /// can run a module declaring `X.Y` if and only if `X == A && Y <= B`.
 /// So, the minor version is intended for backwards-compatible changes, e.g. adding a new function,
 /// and the major version is for fully breaking changes.
-pub const ABI_VERSION: u32 = 0x0004_0000;
+pub const ABI_VERSION: u32 = 0x0005_0000;
 
 /// Provides a raw set of sys calls which abstractions can be built atop of.
 pub mod raw {
@@ -395,10 +395,6 @@ pub mod raw {
             /// Required. id is an index into the `ModuleDef.reducers` returned from `__describe_module__`.
             /// args is a bsatn-encoded product value defined by the schema at `reducers[id]`.
             fn __call_reducer__(id: usize, sender: Identity, timestamp: Timestamp, args: Buffer) -> Result;
-            /// Optional. Called when a client connects to the database.
-            fn __identity_connected__(sender: Identity, timestamp: Timestamp) -> Result;
-            /// Optional. Called when a client disconnects to the database.
-            fn __identity_disconnected__(sender: Identity, timestamp: Timestamp) -> Result;
             /// Currently unused?
             fn __migrate_database__XXXX(sender: Identity, timestamp: Timestamp, something: Buffer) -> Result;
         }

crates/bindings/src/rt.rs

@@ -67,20 +67,6 @@ pub fn create_index(index_name: &str, table_id: u32, index_type: sys::raw::Index
     cvt_result(result.map_err(cvt_errno))
 }
 
-/// Runs the on-connect function `f` provided with a new reducer context
-/// created from `sender` and `timestamp`.
-pub fn invoke_connection_func<R: ReducerResult>(
-    f: impl Fn(ReducerContext) -> R,
-    sender: Buffer,
-    client_address: Buffer,
-    timestamp: u64,
-) -> Buffer {
-    let ctx = assemble_context(sender, timestamp, client_address);
-
-    let res = with_timestamp_set(ctx.timestamp, || f(ctx).into_result());
-    cvt_result(res)
-}
-
 /// Creates a reducer context from the given `sender`, `timestamp` and `client_address`.
 ///
 /// `sender` must contain 32 bytes, from which we will read an `Identity`.
@@ -205,9 +191,9 @@ pub trait ReducerArg<'de> {}
 impl<'de, T: Deserialize<'de>> ReducerArg<'de> for T {}
 impl ReducerArg<'_> for ReducerContext {}
 /// Assert that `T: ReducerArg`.
-pub fn assert_reducerarg<'de, T: ReducerArg<'de>>() {}
+pub fn assert_reducer_arg<'de, T: ReducerArg<'de>>() {}
 /// Assert that `T: ReducerResult`.
-pub fn assert_reducerret<T: ReducerResult>() {}
+pub fn assert_reducer_ret<T: ReducerResult>() {}
 /// Assert that `T: TableType`.
 pub const fn assert_table<T: TableType>() {}