//! `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::authz::{self, AuthzRepo, Capability}; use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxSourceKind}; use crate::repo::ScriptRepository; pub struct InvokeServiceImpl { scripts: Arc, routes: Arc, outbox: Arc, /// F-S-012: optional. When set, invoke()/invoke_async() require /// AppInvoke for authenticated callers; anonymous callers continue /// to skip the check under the script-as-gate convention. authz: Option>, } impl InvokeServiceImpl { #[must_use] pub fn new( scripts: Arc, routes: Arc, outbox: Arc, ) -> Self { Self { scripts, routes, outbox, authz: None, } } /// F-S-012: attach the authz repo so authenticated callers get /// gated on AppInvoke. Without this builder call, the service is /// open to any same-app script (the previous behaviour). #[must_use] pub fn with_authz(mut self, authz: Arc) -> Self { self.authz = Some(authz); self } async fn check_invoke(&self, cx: &SdkCallCx) -> Result<(), InvokeError> { let Some(authz_repo) = self.authz.as_ref() else { return Ok(()); }; authz::script_gate( authz_repo.as_ref(), cx, Capability::AppInvoke(cx.app_id), || InvokeError::Forbidden, InvokeError::Backend, ) .await } async fn resolve_id( &self, cx: &SdkCallCx, script_id: ScriptId, ) -> Result { let script = self .scripts .get(script_id) .await .map_err(|e| InvokeError::Backend(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::Backend(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 { // F-S-012: AppInvoke gate (skipped for anonymous callers). self.check_invoke(cx).await?; 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 { // F-S-012: gate via resolve() which now calls check_invoke first. 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::Backend(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; } }