Allow component functions to be called from node actions (#29821)

This PR adds support for calling component queries, mutations, actions, and scheduling component functions from node actions.

GitOrigin-RevId: 81f1f22055b6ae0911144fbcb7f2a5b6e82e8cc7

Author: emmaling27

crates/application/src/lib.rs

@@ -43,6 +43,8 @@ use common::{
         ComponentId,
         ComponentPath,
         PublicFunctionPath,
+        Reference,
+        Resource,
     },
     document::{
         DocumentUpdate,
@@ -167,6 +169,7 @@ use model::{
         config::ComponentConfigModel,
         handles::FunctionHandlesModel,
         types::ProjectConfig,
+        ComponentsModel,
     },
     config::{
         module_loader::ModuleLoader,
@@ -2755,6 +2758,41 @@ impl<RT: Runtime> Application<RT> {
         FunctionHandlesModel::new(&mut tx).lookup(handle).await
     }
 
+    pub async fn canonicalized_function_path(
+        &self,
+        identity: Identity,
+        component_id: ComponentId,
+        path: Option<String>,
+        reference: Option<String>,
+    ) -> anyhow::Result<CanonicalizedComponentFunctionPath> {
+        let reference = match (path, reference) {
+            (None, None) => anyhow::bail!(ErrorMetadata::bad_request(
+                "MissingUdfPathOrFunctionReference",
+                "Missing UDF path or function reference. One must be provided."
+            )),
+            (Some(path), None) => Reference::Function(path.parse()?),
+            (None, Some(reference)) => reference.parse()?,
+            (Some(_), Some(_)) => anyhow::bail!(ErrorMetadata::bad_request(
+                "InvalidUdfPathOrFunctionReference",
+                "Both UDF path and function reference provided. Only one must be provided."
+            )),
+        };
+        // Reading from a separate transaction here is safe because the component id to
+        // component path mapping is stable.
+        let mut tx = self.begin(identity).await?;
+        let mut components_model = ComponentsModel::new(&mut tx);
+        let resource = components_model
+            .resolve(component_id, None, &reference)
+            .await?;
+        let path = match resource {
+            Resource::Function(path) => path,
+            Resource::Value(_) | Resource::ResolvedSystemUdf(_) => {
+                anyhow::bail!("Resource type not supported for internal query")
+            },
+        };
+        Ok(path)
+    }
+
     pub fn files_storage(&self) -> Arc<dyn Storage> {
         self.files_storage.clone()
     }

crates/local_backend/src/node_action_callbacks.rs

@@ -16,10 +16,8 @@ use axum::{
 };
 use common::{
     components::{
-        CanonicalizedComponentFunctionPath,
         ComponentFunctionPath,
         ComponentId,
-        ComponentPath,
         PublicFunctionPath,
     },
     execution_context::{
@@ -77,15 +75,23 @@ use vector::{
 
 use crate::{
     authentication::ExtractAuthenticationToken,
-    parse::parse_udf_path,
     public_api::{
         export_value,
-        UdfPostRequest,
         UdfResponse,
     },
     LocalAppState,
 };
 
+#[derive(Deserialize, Debug)]
+#[serde(rename_all = "camelCase")]
+pub struct NodeCallbackUdfPostRequest {
+    pub path: Option<String>,
+    pub reference: Option<String>,
+    pub args: UdfArgsJson,
+
+    pub format: Option<String>,
+}
+
 /// This is like `public_query_post`, except it allows calling internal
 /// functions as well. This should not be used for any publicly accessible
 /// endpoints, and should only be used to support Convex functions calling into
@@ -96,21 +102,21 @@ pub async fn internal_query_post(
     State(st): State<LocalAppState>,
     ExtractActionIdentity {
         identity,
-        component_id: _,
+        component_id,
     }: ExtractActionIdentity,
     ExtractClientVersion(client_version): ExtractClientVersion,
     ExtractExecutionContext(context): ExtractExecutionContext,
-    Json(req): Json<UdfPostRequest>,
+    Json(req): Json<NodeCallbackUdfPostRequest>,
 ) -> Result<impl IntoResponse, HttpResponseError> {
-    let udf_path = parse_udf_path(&req.path)?;
+    let path = st
+        .application
+        .canonicalized_function_path(identity.clone(), component_id, req.path, req.reference)
+        .await?;
     let udf_return = st
         .application
         .read_only_udf(
             context.request_id,
-            PublicFunctionPath::Component(CanonicalizedComponentFunctionPath {
-                component: ComponentPath::TODO(),
-                udf_path,
-            }),
+            PublicFunctionPath::Component(path),
             req.args.into_arg_vec(),
             identity,
             FunctionCaller::Action {
@@ -144,21 +150,21 @@ pub async fn internal_mutation_post(
     State(st): State<LocalAppState>,
     ExtractActionIdentity {
         identity,
-        component_id: _,
+        component_id,
     }: ExtractActionIdentity,
     ExtractClientVersion(client_version): ExtractClientVersion,
     ExtractExecutionContext(context): ExtractExecutionContext,
-    Json(req): Json<UdfPostRequest>,
+    Json(req): Json<NodeCallbackUdfPostRequest>,
 ) -> Result<impl IntoResponse, HttpResponseError> {
-    let udf_path = parse_udf_path(&req.path)?;
+    let path = st
+        .application
+        .canonicalized_function_path(identity.clone(), component_id, req.path, req.reference)
+        .await?;
     let udf_result = st
         .application
         .mutation_udf(
             context.request_id,
-            PublicFunctionPath::Component(CanonicalizedComponentFunctionPath {
-                component: ComponentPath::TODO(),
-                udf_path,
-            }),
+            PublicFunctionPath::Component(path),
             req.args.into_arg_vec(),
             identity,
             None,
@@ -197,21 +203,21 @@ pub async fn internal_action_post(
     State(st): State<LocalAppState>,
     ExtractActionIdentity {
         identity,
-        component_id: _,
+        component_id,
     }: ExtractActionIdentity,
     ExtractClientVersion(client_version): ExtractClientVersion,
     ExtractExecutionContext(context): ExtractExecutionContext,
-    Json(req): Json<UdfPostRequest>,
+    Json(req): Json<NodeCallbackUdfPostRequest>,
 ) -> Result<impl IntoResponse, HttpResponseError> {
-    let udf_path = parse_udf_path(&req.path)?;
+    let path = st
+        .application
+        .canonicalized_function_path(identity.clone(), component_id, req.path, req.reference)
+        .await?;
     let udf_result = st
         .application
         .action_udf(
             context.request_id,
-            PublicFunctionPath::Component(CanonicalizedComponentFunctionPath {
-                component: ComponentPath::TODO(),
-                udf_path,
-            }),
+            PublicFunctionPath::Component(path),
             req.args.into_arg_vec(),
             identity,
             FunctionCaller::Action {
@@ -241,7 +247,8 @@ pub async fn internal_action_post(
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ScheduleJobRequest {
-    udf_path: String,
+    reference: Option<String>,
+    udf_path: Option<String>,
     udf_args: UdfArgsJson,
     scheduled_ts: f64,
 }
@@ -264,20 +271,21 @@ pub async fn schedule_job(
 ) -> Result<impl IntoResponse, HttpResponseError> {
     let scheduled_ts = UnixTimestamp::from_secs_f64(req.scheduled_ts);
     // User might have entered an invalid path, so this is a developer error.
-    let udf_path = req.udf_path.parse::<CanonicalizedUdfPath>().map_err(|e| {
-        anyhow::anyhow!(ErrorMetadata::bad_request("InvalidUdfPath", e.to_string()))
-    })?;
+    let path = st
+        .application
+        .canonicalized_function_path(identity.clone(), component_id, req.udf_path, req.reference)
+        .await
+        .map_err(|e| {
+            anyhow::anyhow!(ErrorMetadata::bad_request("InvalidUdfPath", e.to_string()))
+        })?;
     let udf_args = req.udf_args.into_arg_vec();
     let job_id = st
         .application
         .runner()
         .schedule_job(
             identity,
             component_id,
-            CanonicalizedComponentFunctionPath {
-                component: ComponentPath::TODO(),
-                udf_path,
-            },
+            path,
             udf_args,
             scheduled_ts,
             context,

crates/local_backend/src/public_api.rs

@@ -51,7 +51,7 @@ use crate::{
     RouterState,
 };
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct UdfPostRequest {
     pub path: String,

npm-packages/convex/src/server/impl/actions_impl.ts

@@ -22,7 +22,7 @@ function syscallArgs(
 export function getFunctionAddress(functionReference: any) {
   // The `run*` syscalls expect either a UDF path at "name" or a serialized
   // reference at "reference". Dispatch on `functionReference` to coerce
-  // it to one ore the other.
+  // it to one or the other.
   let functionAddress;
 
   // Legacy path for passing in UDF paths directly as function references.

npm-packages/node-executor/src/syscalls.ts

@@ -14,7 +14,8 @@ const STATUS_CODE_BAD_REQUEST = 400;
 const STATUS_CODE_UDF_FAILED = 560;
 
 const runFunctionArgs = z.object({
-  name: z.string(),
+  name: z.optional(z.string()),
+  reference: z.optional(z.string()),
   args: z.any(),
   version: z.string(),
 });
@@ -32,7 +33,8 @@ const runFunctionReturn = z.union([
 ]);
 
 const scheduleSchema = z.object({
-  name: z.string(),
+  name: z.optional(z.string()),
+  reference: z.optional(z.string()),
   ts: z.number(),
   args: z.any(),
   version: z.string(),
@@ -181,7 +183,9 @@ export class SyscallsImpl {
       const parsedArgs = argValidator.parse(args);
       return parsedArgs;
     } catch (e) {
-      throw new Error(`Invalid ${operationName} request`);
+      throw new Error(
+        `Invalid ${operationName} request with args ${JSON.stringify(args)}`,
+      );
     }
   }
 
@@ -354,7 +358,11 @@ export class SyscallsImpl {
     };
     const queryResult = await this.actionCallback({
       version: queryArgs.version,
-      body: { path: queryArgs.name, args: queryArgs.args },
+      body: {
+        path: queryArgs.name,
+        reference: queryArgs.reference,
+        args: queryArgs.args,
+      },
       path: "/api/actions/query",
       operationName,
       responseValidator: runFunctionReturn,
@@ -391,7 +399,11 @@ export class SyscallsImpl {
     };
     const mutationResult = await this.actionCallback({
       version: mutationArgs.version,
-      body: { path: mutationArgs.name, args: mutationArgs.args },
+      body: {
+        path: mutationArgs.name,
+        reference: mutationArgs.reference,
+        args: mutationArgs.args,
+      },
       path: "/api/actions/mutation",
       operationName,
       responseValidator: runFunctionReturn,
@@ -428,7 +440,11 @@ export class SyscallsImpl {
     };
     const actionResult = await this.actionCallback({
       version: actionArgs.version,
-      body: { path: actionArgs.name, args: actionArgs.args },
+      body: {
+        path: actionArgs.name,
+        reference: actionArgs.reference,
+        args: actionArgs.args,
+      },
       path: "/api/actions/action",
       operationName,
       responseValidator: runFunctionReturn,
@@ -486,6 +502,7 @@ export class SyscallsImpl {
     const { jobId } = await this.actionCallback({
       version: scheduleArgs.version,
       body: {
+        reference: scheduleArgs.reference,
         udfPath: scheduleArgs.name,
         udfArgs: scheduleArgs.args,
         scheduledTs: scheduleArgs.ts,