Basic procedure calling

Author: coolreader18

crates/bindings-typescript/src/lib/procedures.ts

@@ -0,0 +1,80 @@
+import { AlgebraicType, ProductType } from '../lib/algebraic_type';
+import BinaryReader from '../lib/binary_reader';
+import BinaryWriter from '../lib/binary_writer';
+import type { ConnectionId } from '../lib/connection_id';
+import type { Identity } from '../lib/identity';
+import type { Timestamp } from '../lib/timestamp';
+import type { ParamsObj } from './reducers';
+import { MODULE_DEF, type UntypedSchemaDef } from './schema';
+import type { Infer, InferTypeOfRow, TypeBuilder } from './type_builders';
+import { bsatnBaseSize } from './util';
+
+export type ProcedureFn<
+  S extends UntypedSchemaDef,
+  Params extends ParamsObj,
+  Ret extends TypeBuilder<any, any>,
+> = (ctx: ProcedureCtx<S>, args: InferTypeOfRow<Params>) => Infer<Ret>;
+
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
+export type ProcedureCtx<S extends UntypedSchemaDef> = Readonly<{
+  sender: Identity;
+  timestamp: Timestamp;
+  connectionId: ConnectionId | null;
+}>;
+
+export function procedure<
+  S extends UntypedSchemaDef,
+  Params extends ParamsObj,
+  Ret extends TypeBuilder<any, any>,
+>(name: string, params: Params, ret: Ret, fn: ProcedureFn<S, Params, Ret>) {
+  const paramsType: ProductType = {
+    elements: Object.entries(params).map(([n, c]) => ({
+      name: n,
+      algebraicType:
+        'typeBuilder' in c ? c.typeBuilder.algebraicType : c.algebraicType,
+    })),
+  };
+  const returnType = ret.algebraicType;
+
+  MODULE_DEF.miscExports.push({
+    tag: 'Procedure',
+    value: {
+      name,
+      params: paramsType,
+      returnType,
+    },
+  });
+
+  PROCEDURES.push({
+    fn,
+    paramsType,
+    returnType,
+    returnTypeBaseSize: bsatnBaseSize(MODULE_DEF.typespace, returnType),
+  });
+}
+
+const PROCEDURES: Array<{
+  fn: ProcedureFn<any, any, any>;
+  paramsType: ProductType;
+  returnType: AlgebraicType;
+  returnTypeBaseSize: number;
+}> = [];
+
+export function callProcedure(
+  id: number,
+  sender: Identity,
+  connectionId: ConnectionId | null,
+  timestamp: Timestamp,
+  argsBuf: Uint8Array
+): Uint8Array {
+  const { fn, paramsType, returnType, returnTypeBaseSize } = PROCEDURES[id];
+  const args = ProductType.deserializeValue(
+    new BinaryReader(argsBuf),
+    paramsType,
+    MODULE_DEF.typespace
+  );
+  const ret = fn(Object.freeze({ sender, connectionId, timestamp }), args);
+  const retBuf = new BinaryWriter(returnTypeBaseSize);
+  AlgebraicType.serializeValue(retBuf, returnType, ret, MODULE_DEF.typespace);
+  return retBuf.getBuffer();
+}

crates/bindings-typescript/src/lib/schema.ts

@@ -47,6 +47,7 @@ import {
 } from './views';
 import RawIndexDefV9 from './autogen/raw_index_def_v_9_type';
 import type { IndexOpts } from './indexes';
+import { procedure, type ProcedureFn } from './procedures';
 
 export type TableNamesOf<S extends UntypedSchemaDef> =
   S['tables'][number]['name'];
@@ -547,6 +548,32 @@ class Schema<S extends UntypedSchemaDef> {
   //   }
   // }
 
+  procedure<Params extends ParamsObj, Ret extends TypeBuilder<any, any>>(
+    name: string,
+    params: Params,
+    ret: Ret,
+    fn: ProcedureFn<S, Params, Ret>
+  ): ProcedureFn<S, Params, Ret>;
+  procedure<Ret extends TypeBuilder<any, any>>(
+    name: string,
+    ret: Ret,
+    fn: ProcedureFn<S, {}, Ret>
+  ): ProcedureFn<S, {}, Ret>;
+  procedure<Params extends ParamsObj, Ret extends TypeBuilder<any, any>>(
+    name: string,
+    paramsOrRet: Ret | Params,
+    retOrFn: ProcedureFn<S, {}, Ret> | Ret,
+    maybeFn?: ProcedureFn<S, Params, Ret>
+  ): ProcedureFn<S, Params, Ret> {
+    if (typeof retOrFn === 'function') {
+      procedure(name, {}, paramsOrRet as Ret, retOrFn);
+      return retOrFn;
+    } else {
+      procedure(name, paramsOrRet as Params, retOrFn, maybeFn!);
+      return maybeFn!;
+    }
+  }
+
   clientVisibilityFilter = {
     sql(filter: string): void {
       MODULE_DEF.rowLevelSecurity.push({ sql: filter });

crates/bindings-typescript/src/server/runtime.ts

@@ -1,4 +1,4 @@
-import { AlgebraicType } from '../lib/algebraic_type';
+import { AlgebraicType, ProductType } from '../lib/algebraic_type';
 import RawModuleDef from '../lib/autogen/raw_module_def_type';
 import type RawModuleDefV9 from '../lib/autogen/raw_module_def_v_9_type';
 import type RawTableDefV9 from '../lib/autogen/raw_table_def_v_9_type';
@@ -38,6 +38,7 @@ import {
   type AnonymousViewCtx,
   type ViewCtx,
 } from '../lib/views';
+import { callProcedure } from '../lib/procedures';
 
 const { freeze } = Object;
 
@@ -235,9 +236,9 @@ export const hooks_v1_1: import('spacetime:sys@1.1').ModuleHooks = {
       // at runtime
       db: getDbView(),
     });
-    const args = AlgebraicType.deserializeValue(
+    const args = ProductType.deserializeValue(
       new BinaryReader(argsBuf),
-      AlgebraicType.Product(params),
+      params,
       MODULE_DEF.typespace
     );
     const ret = fn(ctx, args);
@@ -253,9 +254,9 @@ export const hooks_v1_1: import('spacetime:sys@1.1').ModuleHooks = {
       // at runtime
       db: getDbView(),
     });
-    const args = AlgebraicType.deserializeValue(
+    const args = ProductType.deserializeValue(
       new BinaryReader(argsBuf),
-      AlgebraicType.Product(params),
+      params,
       MODULE_DEF.typespace
     );
     const ret = fn(ctx, args);
@@ -265,6 +266,18 @@ export const hooks_v1_1: import('spacetime:sys@1.1').ModuleHooks = {
   },
 };
 
+export const hooks_v1_2: import('spacetime:sys@1.2').ModuleHooks = {
+  __call_procedure__(id, sender, connection_id, timestamp, args) {
+    return callProcedure(
+      id,
+      new Identity(sender),
+      ConnectionId.nullIfZero(new ConnectionId(connection_id)),
+      new Timestamp(timestamp),
+      args
+    );
+  },
+};
+
 let DB_VIEW: DbView<any> | null = null;
 function getDbView() {
   DB_VIEW ??= makeDbView(MODULE_DEF);

crates/bindings-typescript/src/server/sys.d.ts

@@ -75,3 +75,17 @@ declare module 'spacetime:sys@1.1' {
 
   export function register_hooks(hooks: ModuleHooks);
 }
+
+declare module 'spacetime:sys@1.2' {
+  export type ModuleHooks = {
+    __call_procedure__(
+      id: u32,
+      sender: u256,
+      connection_id: u128,
+      timestamp: bigint,
+      args: Uint8Array
+    ): Uint8Array;
+  };
+
+  export function register_hooks(hooks: ModuleHooks);
+}

crates/core/src/host/module_host.rs

@@ -400,13 +400,6 @@ impl Instance {
             Instance::Js(inst) => inst.trapped(),
         }
     }
-
-    async fn call_procedure(&mut self, params: CallProcedureParams) -> Result<ProcedureCallResult, ProcedureCallError> {
-        match self {
-            Instance::Wasm(inst) => inst.call_procedure(params).await,
-            Instance::Js(inst) => inst.call_procedure(params).await,
-        }
-    }
 }
 
 /// Creates the table for `table_def` in `stdb`.
@@ -1595,18 +1588,20 @@ impl ModuleHost {
         let args = args.into_tuple(procedure_seed).map_err(InvalidProcedureArguments)?;
         let caller_connection_id = caller_connection_id.unwrap_or(ConnectionId::ZERO);
 
-        self.call_async_with_instance(&procedure_def.name, async move |mut inst| {
-            let res = inst
-                .call_procedure(CallProcedureParams {
-                    timestamp: Timestamp::now(),
-                    caller_identity,
-                    caller_connection_id,
-                    timer,
-                    procedure_id,
-                    args,
-                })
-                .await;
-            (res, inst)
+        let params = CallProcedureParams {
+            timestamp: Timestamp::now(),
+            caller_identity,
+            caller_connection_id,
+            timer,
+            procedure_id,
+            args,
+        };
+        self.call_async_with_instance(&procedure_def.name, async move |inst| match inst {
+            Instance::Wasm(mut inst) => (inst.call_procedure(params).await, Instance::Wasm(inst)),
+            Instance::Js(inst) => {
+                let (r, s) = inst.call_procedure(params).await;
+                (r, Instance::Js(s))
+            }
         })
         .await?
     }

crates/core/src/host/v8/mod.rs

@@ -6,8 +6,8 @@ use self::error::{
 use self::ser::serialize_to_js;
 use self::string::{str_from_ident, IntoJsString};
 use self::syscall::{
-    call_call_reducer, call_call_view, call_call_view_anon, call_describe_module, get_hooks, resolve_sys_module, FnRet,
-    HookFunctions,
+    call_call_procedure, call_call_reducer, call_call_view, call_call_view_anon, call_describe_module, get_hooks,
+    resolve_sys_module, FnRet, HookFunctions,
 };
 use super::module_common::{build_common_module_from_raw, run_describer, ModuleCommon};
 use super::module_host::{CallProcedureParams, CallReducerParams, Module, ModuleInfo, ModuleRuntime};
@@ -26,14 +26,17 @@ use crate::host::wasm_common::module_host_actor::{
     WasmInstance,
 };
 use crate::host::wasm_common::{RowIters, TimingSpanSet};
-use crate::host::{ModuleHost, ReducerCallError, ReducerCallResult, Scheduler};
+use crate::host::{
+    ModuleHost, ProcedureCallError, ProcedureCallResult, ReducerCallError, ReducerCallResult, Scheduler,
+};
 use crate::module_host_context::{ModuleCreationContext, ModuleCreationContextLimited};
 use crate::replica_context::ReplicaContext;
 use crate::util::asyncify;
 use anyhow::Context as _;
 use core::any::type_name;
 use core::str;
 use enum_as_inner::EnumAsInner;
+use futures::FutureExt;
 use itertools::Either;
 use spacetimedb_auth::identity::ConnectionAuthCtx;
 use spacetimedb_client_api_messages::energy::FunctionBudget;
@@ -381,10 +384,16 @@ impl JsInstance {
     }
 
     pub async fn call_procedure(
-        &mut self,
-        _params: CallProcedureParams,
-    ) -> Result<super::ProcedureCallResult, super::ProcedureCallError> {
-        todo!("JS/TS module procedure support")
+        self: Box<Self>,
+        params: CallProcedureParams,
+    ) -> (Result<ProcedureCallResult, ProcedureCallError>, Box<Self>) {
+        let (r, s) = self
+            .send_recv(
+                JsWorkerReply::into_call_procedure,
+                JsWorkerRequest::CallProcedure { params },
+            )
+            .await;
+        (*r, s)
     }
 
     pub async fn call_view(self: Box<Self>, tx: MutTxId, params: CallViewParams) -> (ViewCallResult, Box<Self>) {
@@ -401,6 +410,7 @@ enum JsWorkerReply {
     UpdateDatabase(anyhow::Result<UpdateDatabaseResult>),
     CallReducer(ReducerCallResult),
     CallView(Box<ViewCallResult>),
+    CallProcedure(Box<Result<ProcedureCallResult, ProcedureCallError>>),
     ClearAllClients(anyhow::Result<()>),
     CallIdentityConnected(Result<(), ClientConnectedError>),
     CallIdentityDisconnected(Result<(), ReducerCallError>),
@@ -426,6 +436,8 @@ enum JsWorkerRequest {
     },
     /// See [`JsInstance::call_view`].
     CallView { tx: MutTxId, params: CallViewParams },
+    /// See [`JsInstance::call_procedure`].
+    CallProcedure { params: CallProcedureParams },
     /// See [`JsInstance::clear_all_clients`].
     ClearAllClients,
     /// See [`JsInstance::call_identity_connected`].
@@ -579,6 +591,14 @@ fn spawn_instance_worker(
                     let (res, trapped) = instance_common.call_view_with_tx(tx, params, &mut inst);
                     reply("call_view", JsWorkerReply::CallView(res.into()), trapped);
                 }
+                JsWorkerRequest::CallProcedure { params } => {
+                    let (res, trapped) = instance_common
+                        .call_procedure(params, &mut inst)
+                        .now_or_never()
+                        .expect("our call_procedure implementation is not actually async");
+
+                    reply("call_procedure", JsWorkerReply::CallProcedure(res.into()), trapped);
+                }
                 JsWorkerRequest::ClearAllClients => {
                     let res = instance_common.clear_all_clients();
                     reply("clear_all_clients", ClearAllClients(res), false);
@@ -758,11 +778,10 @@ impl WasmInstance for V8Instance<'_, '_, '_> {
     }
 
     fn call_reducer(&mut self, op: ReducerOp<'_>, budget: FunctionBudget) -> ReducerExecuteResult {
-        let ExecutionResult { stats, call_result } = common_call(self.scope, budget, op, |scope, op| {
+        common_call(self.scope, budget, op, |scope, op| {
             Ok(call_call_reducer(scope, self.hooks, op)?)
-        });
-        let call_result = call_result.and_then(|res| res.map_err(ExecutionError::User));
-        ExecutionResult { stats, call_result }
+        })
+        .map_result(|call_result| call_result.and_then(|res| res.map_err(ExecutionError::User)))
     }
 
     fn call_view(&mut self, op: ViewOp<'_>, budget: FunctionBudget) -> ViewExecuteResult {
@@ -781,8 +800,16 @@ impl WasmInstance for V8Instance<'_, '_, '_> {
         log_traceback(self.replica_ctx, func_type, func, trap)
     }
 
-    async fn call_procedure(&mut self, _op: ProcedureOp, _budget: FunctionBudget) -> ProcedureExecuteResult {
-        todo!("JS/TS module procedure support")
+    async fn call_procedure(&mut self, op: ProcedureOp, budget: FunctionBudget) -> ProcedureExecuteResult {
+        common_call(self.scope, budget, op, |scope, op| {
+            call_call_procedure(scope, self.hooks, op)
+        })
+        .map_result(|call_result| {
+            call_result.map_err(|e| match e {
+                ExecutionError::User(e) => anyhow::Error::msg(e),
+                ExecutionError::Recoverable(e) | ExecutionError::Trap(e) => e,
+            })
+        })
     }
 }
 
@@ -812,7 +839,7 @@ where
             CanContinue::Yes => ExecutionError::Recoverable(e.into()),
             CanContinue::YesCancelTermination => {
                 scope.cancel_terminate_execution();
-                ExecutionError::Trap(e.into())
+                ExecutionError::Recoverable(e.into())
             }
         }
     });