PiCloud/crates/manager-core/src/invoke_service.rs

//! `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<dyn ScriptRepository>,
    routes: Arc<RouteTable>,
    outbox: Arc<dyn OutboxRepo>,
    /// 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<Arc<dyn AuthzRepo>>,
}

impl InvokeServiceImpl {
    #[must_use]
    pub fn new(
        scripts: Arc<dyn ScriptRepository>,
        routes: Arc<RouteTable>,
        outbox: Arc<dyn OutboxRepo>,
    ) -> 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<dyn AuthzRepo>) -> 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<ResolvedScript, InvokeError> {
        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<ResolvedScript, InvokeError> {
        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<ResolvedScript, InvokeError> {
        // 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<ResolvedScript, InvokeError> {
        // 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<ExecutionId, InvokeError> {
        // 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<Option<Script>, 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<Option<Script>, ScriptRepositoryError> {
            Ok(
                if app_id == self.script.app_id && name == self.script.name {
                    Some(self.script.clone())
                } else {
                    None
                },
            )
        }
        async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError> {
            Ok(vec![self.script.clone()])
        }
        async fn list_for_app(&self, _app_id: AppId) -> Result<Vec<Script>, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn list_for_user(
            &self,
            _user_id: AdminUserId,
        ) -> Result<Vec<Script>, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn create(&self, _input: NewScript) -> Result<Script, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn update(
            &self,
            _id: ScriptId,
            _patch: ScriptPatch,
        ) -> Result<Script, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn delete(&self, _id: ScriptId) -> Result<(), ScriptRepositoryError> {
            unimplemented!()
        }
        async fn count_routes_for_script(
            &self,
            _script_id: ScriptId,
        ) -> Result<i64, ScriptRepositoryError> {
            Ok(0)
        }
        async fn count_triggers_for_script(
            &self,
            _script_id: ScriptId,
        ) -> Result<i64, ScriptRepositoryError> {
            Ok(0)
        }
        async fn list_imports(
            &self,
            _script_id: ScriptId,
        ) -> Result<Vec<Script>, ScriptRepositoryError> {
            Ok(vec![])
        }
    }

    struct CapturingOutbox {
        last: tokio::sync::Mutex<Option<NewOutboxRow>>,
    }
    #[async_trait]
    impl OutboxRepo for CapturingOutbox {
        async fn insert(&self, row: NewOutboxRow) -> Result<uuid::Uuid, OutboxRepoError> {
            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<Vec<OutboxRow>, 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<Utc>,
        ) -> 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;
    }
}