diff --git a/crates/manager-core/src/invoke_service.rs b/crates/manager-core/src/invoke_service.rs new file mode 100644 index 0000000..99fcbb8 --- /dev/null +++ b/crates/manager-core/src/invoke_service.rs @@ -0,0 +1,412 @@ +//! `InvokeServiceImpl` — wires `ScriptRepository` + `RouteTable` + +//! `OutboxRepo` underneath the `picloud_shared::InvokeService` trait. +//! +//! Same-app only. The cross-app guard lives at the service entry point: +//! every resolved script's `app_id` must match `cx.app_id` — otherwise +//! `InvokeError::CrossApp` (v1.1.x maintains strict isolation). +//! +//! `invoke_async` writes a v1.1.9 `OutboxSourceKind::Invoke` row the +//! dispatcher consumes (commit 10). Runs once per row — no retries. +//! Callers wrap in `retry::with(...)` if they want retries. + +use std::sync::Arc; + +use async_trait::async_trait; +use picloud_orchestrator_core::routing::{matcher, RouteTable}; +use picloud_shared::{ + ExecutionId, InvokeError, InvokeService, InvokeTarget, ResolvedScript, ScriptId, SdkCallCx, +}; + +use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxSourceKind}; +use crate::repo::ScriptRepository; + +pub struct InvokeServiceImpl { + scripts: Arc, + routes: Arc, + outbox: Arc, +} + +impl InvokeServiceImpl { + #[must_use] + pub fn new( + scripts: Arc, + routes: Arc, + outbox: Arc, + ) -> Self { + Self { + scripts, + routes, + outbox, + } + } + + async fn resolve_id( + &self, + cx: &SdkCallCx, + script_id: ScriptId, + ) -> Result { + let script = self + .scripts + .get(script_id) + .await + .map_err(|e| InvokeError::Unavailable(e.to_string()))? + .ok_or_else(|| InvokeError::NotFound(format!("id {script_id}")))?; + if script.app_id != cx.app_id { + return Err(InvokeError::CrossApp); + } + Ok(ResolvedScript { + script_id: script.id, + app_id: script.app_id, + source: script.source, + updated_at: script.updated_at, + name: script.name, + }) + } + + async fn resolve_name( + &self, + cx: &SdkCallCx, + name: &str, + ) -> Result { + let script = self + .scripts + .get_by_name(cx.app_id, name) + .await + .map_err(|e| InvokeError::Unavailable(e.to_string()))? + .ok_or_else(|| InvokeError::NotFound(format!("name {name:?}")))?; + Ok(ResolvedScript { + script_id: script.id, + app_id: script.app_id, + source: script.source, + updated_at: script.updated_at, + name: script.name, + }) + } + + async fn resolve_path( + &self, + cx: &SdkCallCx, + path: &str, + ) -> Result { + // Run the matcher against this app's routes only. Host bucket + // is irrelevant for in-app composition — use a synthetic host + // that matches the wildcard tier. Method assumed POST since + // most route-resolution invokes target write endpoints; a more + // exact API would surface method via the SDK in a future bump. + let app_routes = self.routes.snapshot_for_app(cx.app_id); + let m = matcher::r#match(&app_routes, "invoke.local", "POST", path).or_else(|| { + matcher::r#match(&app_routes, "invoke.local", "GET", path) + }); + let m = m.ok_or_else(|| InvokeError::NotFound(format!("path {path:?}")))?; + // Resolve the script the matched route bound. Cross-app + // re-check via the script row (defense in depth — RouteTable + // partitions by app already, but resolve_id verifies again). + self.resolve_id(cx, m.matched.script_id).await + } +} + +#[async_trait] +impl InvokeService for InvokeServiceImpl { + async fn resolve( + &self, + cx: &SdkCallCx, + target: InvokeTarget, + ) -> Result { + match target { + InvokeTarget::Id(id) => self.resolve_id(cx, id).await, + InvokeTarget::Name(n) => self.resolve_name(cx, &n).await, + InvokeTarget::Path(p) => self.resolve_path(cx, &p).await, + } + } + + async fn enqueue_async( + &self, + cx: &SdkCallCx, + target: InvokeTarget, + args: serde_json::Value, + ) -> Result { + let resolved = self.resolve(cx, target).await?; + let execution_id = ExecutionId::new(); + // Payload carries everything the dispatcher's Invoke arm needs + // to build an ExecRequest at fire time. + let payload = serde_json::json!({ + "kind": "invoke", + "script_id": resolved.script_id, + "script_name": resolved.name, + "args": args, + "execution_id": execution_id, + "caller_request_id": cx.request_id, + "root_execution_id": cx.root_execution_id, + "trigger_depth": cx.trigger_depth.saturating_add(1), + }); + // Caller's principal is recorded as origin_principal (forensic), + // but the executing script runs with no principal — invoke_async + // is the fire-and-forget equivalent of a function call, not a + // re-authentication. + let origin_principal = cx.principal.as_ref().map(|p| p.user_id); + self.outbox + .insert(NewOutboxRow { + app_id: cx.app_id, + source_kind: OutboxSourceKind::Invoke, + trigger_id: None, + script_id: Some(resolved.script_id), + reply_to: None, + payload, + origin_principal, + trigger_depth: cx.trigger_depth.saturating_add(1), + root_execution_id: Some(cx.root_execution_id), + }) + .await + .map_err(|e| InvokeError::Unavailable(e.to_string()))?; + Ok(execution_id) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::outbox_repo::{OutboxRepoError, OutboxRow}; + use crate::repo::{NewScript, ScriptPatch, ScriptRepositoryError}; + use chrono::Utc; + use picloud_shared::{AdminUserId, AppId, ExecutionId, RequestId, Script, ScriptSandbox}; + + struct OneScriptRepo { + script: Script, + } + #[async_trait] + impl ScriptRepository for OneScriptRepo { + async fn get( + &self, + id: ScriptId, + ) -> Result, ScriptRepositoryError> { + Ok(if id == self.script.id { + Some(self.script.clone()) + } else { + None + }) + } + async fn get_by_name( + &self, + app_id: AppId, + name: &str, + ) -> Result, ScriptRepositoryError> { + Ok(if app_id == self.script.app_id && name == self.script.name { + Some(self.script.clone()) + } else { + None + }) + } + async fn list(&self) -> Result, ScriptRepositoryError> { + Ok(vec![self.script.clone()]) + } + async fn list_for_app( + &self, + _app_id: AppId, + ) -> Result, ScriptRepositoryError> { + unimplemented!() + } + async fn list_for_user( + &self, + _user_id: AdminUserId, + ) -> Result, ScriptRepositoryError> { + unimplemented!() + } + async fn create(&self, _input: NewScript) -> Result { + unimplemented!() + } + async fn update( + &self, + _id: ScriptId, + _patch: ScriptPatch, + ) -> Result { + unimplemented!() + } + async fn delete(&self, _id: ScriptId) -> Result<(), ScriptRepositoryError> { + unimplemented!() + } + async fn count_routes_for_script( + &self, + _script_id: ScriptId, + ) -> Result { + Ok(0) + } + async fn count_triggers_for_script( + &self, + _script_id: ScriptId, + ) -> Result { + Ok(0) + } + async fn list_imports( + &self, + _script_id: ScriptId, + ) -> Result, ScriptRepositoryError> { + Ok(vec![]) + } + } + + struct CapturingOutbox { + last: tokio::sync::Mutex>, + } + #[async_trait] + impl OutboxRepo for CapturingOutbox { + async fn insert(&self, row: NewOutboxRow) -> Result { + let id = uuid::Uuid::new_v4(); + *self.last.lock().await = Some(row); + Ok(id) + } + async fn claim_due( + &self, + _claimed_by: &str, + _limit: i64, + ) -> Result, OutboxRepoError> { + Ok(vec![]) + } + async fn delete(&self, _id: uuid::Uuid) -> Result<(), OutboxRepoError> { + Ok(()) + } + async fn reschedule( + &self, + _id: uuid::Uuid, + _attempt_count: u32, + _next_attempt_at: chrono::DateTime, + ) -> Result<(), OutboxRepoError> { + Ok(()) + } + } + + fn make_script(app_id: AppId, name: &str) -> Script { + Script { + id: ScriptId::new(), + app_id, + name: name.into(), + description: None, + version: 1, + source: "0".into(), + kind: picloud_shared::ScriptKind::Endpoint, + timeout_seconds: 30, + memory_limit_mb: 64, + sandbox: ScriptSandbox::default(), + created_at: Utc::now(), + updated_at: Utc::now(), + } + } + + fn anon_cx(app_id: AppId) -> SdkCallCx { + let execution_id = ExecutionId::new(); + SdkCallCx { + app_id, + script_id: ScriptId::new(), + principal: None, + execution_id, + request_id: RequestId::new(), + trigger_depth: 0, + root_execution_id: execution_id, + is_dead_letter_handler: false, + event: None, + } + } + + #[tokio::test] + async fn resolve_by_id_same_app_succeeds() { + let app_id = AppId::new(); + let script = make_script(app_id, "worker"); + let svc = InvokeServiceImpl::new( + Arc::new(OneScriptRepo { + script: script.clone(), + }), + Arc::new(RouteTable::new()), + Arc::new(CapturingOutbox { + last: tokio::sync::Mutex::new(None), + }), + ); + let cx = anon_cx(app_id); + let resolved = svc.resolve(&cx, InvokeTarget::Id(script.id)).await.unwrap(); + assert_eq!(resolved.script_id, script.id); + assert_eq!(resolved.app_id, app_id); + } + + #[tokio::test] + async fn resolve_by_id_cross_app_rejects() { + let app_a = AppId::new(); + let app_b = AppId::new(); + let script = make_script(app_a, "worker"); + let svc = InvokeServiceImpl::new( + Arc::new(OneScriptRepo { script: script.clone() }), + Arc::new(RouteTable::new()), + Arc::new(CapturingOutbox { + last: tokio::sync::Mutex::new(None), + }), + ); + // Caller is in app_b but the script belongs to app_a. + let cx = anon_cx(app_b); + let err = svc.resolve(&cx, InvokeTarget::Id(script.id)).await.unwrap_err(); + assert!(matches!(err, InvokeError::CrossApp)); + } + + #[tokio::test] + async fn resolve_by_name_finds_script_in_caller_app() { + let app_id = AppId::new(); + let script = make_script(app_id, "worker"); + let svc = InvokeServiceImpl::new( + Arc::new(OneScriptRepo { script: script.clone() }), + Arc::new(RouteTable::new()), + Arc::new(CapturingOutbox { + last: tokio::sync::Mutex::new(None), + }), + ); + let cx = anon_cx(app_id); + let resolved = svc + .resolve(&cx, InvokeTarget::Name("worker".into())) + .await + .unwrap(); + assert_eq!(resolved.script_id, script.id); + } + + #[tokio::test] + async fn resolve_by_name_unknown_is_not_found() { + let app_id = AppId::new(); + let script = make_script(app_id, "worker"); + let svc = InvokeServiceImpl::new( + Arc::new(OneScriptRepo { script }), + Arc::new(RouteTable::new()), + Arc::new(CapturingOutbox { + last: tokio::sync::Mutex::new(None), + }), + ); + let cx = anon_cx(app_id); + let err = svc + .resolve(&cx, InvokeTarget::Name("ghost".into())) + .await + .unwrap_err(); + assert!(matches!(err, InvokeError::NotFound(_))); + } + + #[tokio::test] + async fn enqueue_async_writes_outbox_row_with_invoke_kind() { + let app_id = AppId::new(); + let script = make_script(app_id, "worker"); + let outbox = Arc::new(CapturingOutbox { + last: tokio::sync::Mutex::new(None), + }); + let svc = InvokeServiceImpl::new( + Arc::new(OneScriptRepo { script: script.clone() }), + Arc::new(RouteTable::new()), + outbox.clone(), + ); + let cx = anon_cx(app_id); + let exec_id = svc + .enqueue_async( + &cx, + InvokeTarget::Id(script.id), + serde_json::json!({ "x": 1 }), + ) + .await + .unwrap(); + let row = outbox.last.lock().await.clone().unwrap(); + assert_eq!(row.app_id, app_id); + assert_eq!(row.source_kind, OutboxSourceKind::Invoke); + assert_eq!(row.script_id, Some(script.id)); + assert_eq!(row.trigger_depth, 1); + let _ = exec_id; + } +} diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 00387d2..5373c3c 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -55,6 +55,7 @@ pub mod outbox_repo; pub mod principal_resolver; pub mod pubsub_repo; pub mod pubsub_service; +pub mod invoke_service; pub mod queue_repo; pub mod queue_service; pub mod realtime_authority; diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 91936b4..10b61ab 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -274,11 +274,27 @@ pub async fn build_app( authz.clone(), ), ); - // v1.1.9 function composition. Real InvokeService wired in commit 8; - // the noop here keeps the binary compiling — without it, scripts - // calling `invoke()` will throw a clear error. - let invoke: Arc = - Arc::new(picloud_shared::NoopInvokeService); + // Route table created early (before Services) so InvokeServiceImpl + // can use it for path resolution. It's populated below from the + // route_repo, then re-populated whenever the admin layer writes + // routes. + let route_table = Arc::new(RouteTable::new()); + let initial = route_repo.list_all().await?; + let compiled = compile_routes(&initial) + .map_err(|e| anyhow::anyhow!("failed to compile stored routes: {e}"))?; + route_table.replace_all(compiled); + + // v1.1.9 function composition. InvokeServiceImpl resolves targets + // by Id, Name (via get_by_name), or Path (via the orchestrator's + // RouteTable). invoke_async writes an OutboxSourceKind::Invoke row + // that the dispatcher fires through the standard executor path. + let invoke: Arc = Arc::new( + picloud_manager_core::invoke_service::InvokeServiceImpl::new( + Arc::new(PostgresScriptRepoHandle(script_repo.clone())), + route_table.clone(), + outbox_repo.clone(), + ), + ); let services = Services::new( kv, docs, @@ -296,13 +312,6 @@ pub async fn build_app( ); let engine = Arc::new(Engine::new(Limits::default(), services)); - // Compile the routes table once at startup; admin writes refresh it. - let route_table = Arc::new(RouteTable::new()); - let initial = route_repo.list_all().await?; - let compiled = compile_routes(&initial) - .map_err(|e| anyhow::anyhow!("failed to compile stored routes: {e}"))?; - route_table.replace_all(compiled); - // Same shape for app domains (Host → app_id cache). let app_domain_table = Arc::new(AppDomainTable::new()); let initial_domains = domains_repo.list_all().await?;