fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(interceptors): extend before/after hooks to docs/files/queue/pubsub/http (§9.4 M7-M11)

Browse Source 
Every non-KV mutating op now runs the same before/after interceptor machinery:
- docs create/update/delete (+ group) — create/update honor the M4 data transform;
- files create/update/delete (+ group) — allow/deny only (the blob is never
  surfaced to the hook, value = None);
- queue enqueue (+ shared) — transform-capable; after reports the message id;
- pubsub publish_durable (+ shared topic) — transform-capable;
- http request — all verbs funnel through two svc.request sites; collection =
  method, key = url, body not surfaced.
ictx threaded into the free-fn/handle paths that lacked it (http was _ictx;
queue/pubsub per-app publish).

validate_bundle_for generalized to a per-service allowed-ops map (kv set/delete;
docs/files create/update/delete; queue enqueue; pubsub publish; http request) —
unknown service/op still rejected. INVARIANT enforced + verified: every allowed
(service, op) has a matching runtime hook, so no validated-but-unhooked
fail-open. Pinned by a new docs-create deny journey (11 interceptor journeys green).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-16 20:37:18 +02:00
parent faf04174c4
commit ef16d48a8c
8 changed files with 564 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
crates/executor-core/src/sdk/docs.rs
View File

@@ -39,9 +39,7 @@ pub struct DocsHandle {
collection: String,
service: Arc<dyn DocsService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing: carried so `create`/`update`/`delete` can run a before/
// after hook in a later milestone (M8). Unused until then.
#[allow(dead_code)]
// §9.4: carried so `create`/`update`/`delete` run the before/after hook (M8).
ictx: InterceptorCtx,
}
@@ -54,8 +52,7 @@ pub struct GroupDocsHandle {
collection: String,
service: Arc<dyn GroupDocsService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing (see `DocsHandle`). Unused until M8.
#[allow(dead_code)]
// §9.4: carried so shared `create`/`update`/`delete` run the hook too (M8).
ictx: InterceptorCtx,
}
@@ -135,12 +132,23 @@ fn register_create(engine: &mut RhaiEngine) {
engine.register_fn(
"create",
|handle: &mut DocsHandle, data: Map| -> Result<String, Box<EvalAltResult>> {
let h = handle.clone();
let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?;
// §9.4 before-op interceptor (allow/deny + M4 data-transform).
let write_val = run_before(handle, "create", "", Some(&json))?.unwrap_or(json);
let written = write_val.clone();
let h = handle.clone();
let id = block_on("docs", async move {
h.service.create(&h.cx, &h.collection, json).await
h.service.create(&h.cx, &h.collection, write_val).await
})?;
Ok(id.to_string())
let id_str = id.to_string();
run_after(
handle,
"create",
"",
Some(&written),
serde_json::Value::String(id_str.clone()),
)?;
Ok(id_str)
},
);
}
@@ -194,14 +202,23 @@ fn register_update(engine: &mut RhaiEngine) {
engine.register_fn(
"update",
|handle: &mut DocsHandle, id: &str, data: Map| -> Result<(), Box<EvalAltResult>> {
let h = handle.clone();
let parsed_id = parse_doc_id(id)?;
let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?;
let write_val = run_before(handle, "update", id, Some(&json))?.unwrap_or(json);
let written = write_val.clone();
let h = handle.clone();
block_on("docs", async move {
h.service
.update(&h.cx, &h.collection, parsed_id, json)
.update(&h.cx, &h.collection, parsed_id, write_val)
.await
})
})?;
run_after(
handle,
"update",
id,
Some(&written),
serde_json::Value::Null,
)
},
);
}
@@ -210,15 +227,62 @@ fn register_delete(engine: &mut RhaiEngine) {
engine.register_fn(
"delete",
|handle: &mut DocsHandle, id: &str| -> Result<bool, Box<EvalAltResult>> {
let h = handle.clone();
let parsed_id = parse_doc_id(id)?;
block_on("docs", async move {
// Delete carries no value, so the before-hook never transforms.
run_before(handle, "delete", id, None)?;
let h = handle.clone();
let was_present = block_on("docs", async move {
h.service.delete(&h.cx, &h.collection, parsed_id).await
})
})?;
run_after(
handle,
"delete",
id,
None,
serde_json::Value::Bool(was_present),
)?;
Ok(was_present)
},
);
}
/// §9.4: run the before-op interceptor for a per-app docs write.
fn run_before(
handle: &DocsHandle,
op: &'static str,
key: &str,
value: Option<&serde_json::Value>,
) -> Result<Option<serde_json::Value>, Box<EvalAltResult>> {
super::interceptor::run_before(
&handle.ictx,
&handle.cx,
"docs",
op,
&handle.collection,
key,
value,
)
}
fn run_after(
handle: &DocsHandle,
op: &'static str,
key: &str,
value: Option<&serde_json::Value>,
result: serde_json::Value,
) -> Result<(), Box<EvalAltResult>> {
super::interceptor::run_after(
&handle.ictx,
&handle.cx,
"docs",
op,
&handle.collection,
key,
value,
result,
)
}
fn register_list(engine: &mut RhaiEngine) {
// Zero-arg form: full page from the start.
engine.register_fn(
@@ -283,12 +347,22 @@ fn register_group_create(engine: &mut RhaiEngine) {
engine.register_fn(
"create",
|handle: &mut GroupDocsHandle, data: Map| -> Result<String, Box<EvalAltResult>> {
let h = handle.clone();
let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?;
let write_val = group_run_before(handle, "create", "", Some(&json))?.unwrap_or(json);
let written = write_val.clone();
let h = handle.clone();
let id = block_on("docs", async move {
h.service.create(&h.cx, &h.collection, json).await
h.service.create(&h.cx, &h.collection, write_val).await
})?;
Ok(id.to_string())
let id_str = id.to_string();
group_run_after(
handle,
"create",
"",
Some(&written),
serde_json::Value::String(id_str.clone()),
)?;
Ok(id_str)
},
);
}
@@ -342,14 +416,23 @@ fn register_group_update(engine: &mut RhaiEngine) {
engine.register_fn(
"update",
|handle: &mut GroupDocsHandle, id: &str, data: Map| -> Result<(), Box<EvalAltResult>> {
let h = handle.clone();
let parsed_id = parse_doc_id(id)?;
let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?;
let write_val = group_run_before(handle, "update", id, Some(&json))?.unwrap_or(json);
let written = write_val.clone();
let h = handle.clone();
block_on("docs", async move {
h.service
.update(&h.cx, &h.collection, parsed_id, json)
.update(&h.cx, &h.collection, parsed_id, write_val)
.await
})
})?;
group_run_after(
handle,
"update",
id,
Some(&written),
serde_json::Value::Null,
)
},
);
}
@@ -358,15 +441,62 @@ fn register_group_delete(engine: &mut RhaiEngine) {
engine.register_fn(
"delete",
|handle: &mut GroupDocsHandle, id: &str| -> Result<bool, Box<EvalAltResult>> {
let h = handle.clone();
let parsed_id = parse_doc_id(id)?;
block_on("docs", async move {
group_run_before(handle, "delete", id, None)?;
let h = handle.clone();
let was_present = block_on("docs", async move {
h.service.delete(&h.cx, &h.collection, parsed_id).await
})
})?;
group_run_after(
handle,
"delete",
id,
None,
serde_json::Value::Bool(was_present),
)?;
Ok(was_present)
},
);
}
/// §9.4: before/after hooks for a shared (group) docs write — mirror the per-app
/// helpers but over `GroupDocsHandle`.
fn group_run_before(
handle: &GroupDocsHandle,
op: &'static str,
key: &str,
value: Option<&serde_json::Value>,
) -> Result<Option<serde_json::Value>, Box<EvalAltResult>> {
super::interceptor::run_before(
&handle.ictx,
&handle.cx,
"docs",
op,
&handle.collection,
key,
value,
)
}
fn group_run_after(
handle: &GroupDocsHandle,
op: &'static str,
key: &str,
value: Option<&serde_json::Value>,
result: serde_json::Value,
) -> Result<(), Box<EvalAltResult>> {
super::interceptor::run_after(
&handle.ictx,
&handle.cx,
"docs",
op,
&handle.collection,
key,
value,
result,
)
}
fn register_group_list(engine: &mut RhaiEngine) {
engine.register_fn(
"list",

140
crates/executor-core/src/sdk/files.rs
View File

@@ -37,9 +37,9 @@ pub struct FilesHandle {
collection: String,
service: Arc<dyn FilesService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing: carried so `create`/`update`/`delete` can run a before/
// after hook in a later milestone (M9). Unused until then.
#[allow(dead_code)]
// §9.4: carried so `create`/`update`/`delete` run the before/after hook (M9).
// Files carry bytes, so the hook sees service/op/collection/key only — the
// blob is never materialized into a `value`, and there is no data transform.
ictx: InterceptorCtx,
}
@@ -52,8 +52,7 @@ pub struct GroupFilesHandle {
collection: String,
service: Arc<dyn GroupFilesService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing (see `FilesHandle`). Unused until M9.
#[allow(dead_code)]
// §9.4: carried so shared `create`/`update`/`delete` run the hook too (M9).
ictx: InterceptorCtx,
}
@@ -133,6 +132,9 @@ fn register_create(engine: &mut RhaiEngine) {
let name = require_string(&meta, "name")?;
let content_type = require_string(&meta, "content_type")?;
let data = require_blob(&meta, "data")?;
// §9.4 before-op interceptor (allow/deny only — no transform for
// files, the blob is not surfaced to the hook).
run_before(handle, "create", "")?;
let h = handle.clone();
let new = NewFile {
name,
@@ -142,7 +144,14 @@ fn register_create(engine: &mut RhaiEngine) {
let id = block_on("files", async move {
h.service.create(&h.cx, &h.collection, new).await
})?;
Ok(id.to_string())
let id_str = id.to_string();
run_after(
handle,
"create",
"",
serde_json::Value::String(id_str.clone()),
)?;
Ok(id_str)
},
);
}
@@ -182,16 +191,18 @@ fn register_update(engine: &mut RhaiEngine) {
let data = require_blob(&meta, "data")?;
let name = optional_string(&meta, "name")?;
let content_type = optional_string(&meta, "content_type")?;
run_before(handle, "update", id)?;
let h = handle.clone();
let id = id.to_string();
let id_owned = id.to_string();
let upd = FileUpdate {
data,
name,
content_type,
};
block_on("files", async move {
h.service.update(&h.cx, &h.collection, &id, upd).await
})
h.service.update(&h.cx, &h.collection, &id_owned, upd).await
})?;
run_after(handle, "update", id, serde_json::Value::Null)
},
);
}
@@ -200,15 +211,52 @@ fn register_delete(engine: &mut RhaiEngine) {
engine.register_fn(
"delete",
|handle: &mut FilesHandle, id: &str| -> Result<bool, Box<EvalAltResult>> {
run_before(handle, "delete", id)?;
let h = handle.clone();
let id = id.to_string();
block_on("files", async move {
h.service.delete(&h.cx, &h.collection, &id).await
})
let id_owned = id.to_string();
let was_present = block_on("files", async move {
h.service.delete(&h.cx, &h.collection, &id_owned).await
})?;
run_after(handle, "delete", id, serde_json::Value::Bool(was_present))?;
Ok(was_present)
},
);
}
/// §9.4: before/after hooks for a per-app files write. Files pass `value = None`
/// (the blob is never surfaced to the hook), so the before-hook only allows or
/// denies — it never transforms.
fn run_before(handle: &FilesHandle, op: &'static str, key: &str) -> Result<(), Box<EvalAltResult>> {
super::interceptor::run_before(
&handle.ictx,
&handle.cx,
"files",
op,
&handle.collection,
key,
None,
)
.map(|_| ())
}
fn run_after(
handle: &FilesHandle,
op: &'static str,
key: &str,
result: serde_json::Value,
) -> Result<(), Box<EvalAltResult>> {
super::interceptor::run_after(
&handle.ictx,
&handle.cx,
"files",
op,
&handle.collection,
key,
None,
result,
)
}
fn register_list(engine: &mut RhaiEngine) {
engine.register_fn(
"list",
@@ -288,6 +336,7 @@ fn register_group_create(engine: &mut RhaiEngine) {
let name = require_string(&meta, "name")?;
let content_type = require_string(&meta, "content_type")?;
let data = require_blob(&meta, "data")?;
group_run_before(handle, "create", "")?;
let h = handle.clone();
let new = NewFile {
name,
@@ -297,7 +346,14 @@ fn register_group_create(engine: &mut RhaiEngine) {
let id = block_on("files", async move {
h.service.create(&h.cx, &h.collection, new).await
})?;
Ok(id.to_string())
let id_str = id.to_string();
group_run_after(
handle,
"create",
"",
serde_json::Value::String(id_str.clone()),
)?;
Ok(id_str)
},
);
}
@@ -337,16 +393,18 @@ fn register_group_update(engine: &mut RhaiEngine) {
let data = require_blob(&meta, "data")?;
let name = optional_string(&meta, "name")?;
let content_type = optional_string(&meta, "content_type")?;
group_run_before(handle, "update", id)?;
let h = handle.clone();
let id = id.to_string();
let id_owned = id.to_string();
let upd = FileUpdate {
data,
name,
content_type,
};
block_on("files", async move {
h.service.update(&h.cx, &h.collection, &id, upd).await
})
h.service.update(&h.cx, &h.collection, &id_owned, upd).await
})?;
group_run_after(handle, "update", id, serde_json::Value::Null)
},
);
}
@@ -355,15 +413,55 @@ fn register_group_delete(engine: &mut RhaiEngine) {
engine.register_fn(
"delete",
|handle: &mut GroupFilesHandle, id: &str| -> Result<bool, Box<EvalAltResult>> {
group_run_before(handle, "delete", id)?;
let h = handle.clone();
let id = id.to_string();
block_on("files", async move {
h.service.delete(&h.cx, &h.collection, &id).await
})
let id_owned = id.to_string();
let was_present = block_on("files", async move {
h.service.delete(&h.cx, &h.collection, &id_owned).await
})?;
group_run_after(handle, "delete", id, serde_json::Value::Bool(was_present))?;
Ok(was_present)
},
);
}
/// §9.4: before/after hooks for a shared (group) files write — mirror the
/// per-app helpers but over `GroupFilesHandle`.
fn group_run_before(
handle: &GroupFilesHandle,
op: &'static str,
key: &str,
) -> Result<(), Box<EvalAltResult>> {
super::interceptor::run_before(
&handle.ictx,
&handle.cx,
"files",
op,
&handle.collection,
key,
None,
)
.map(|_| ())
}
fn group_run_after(
handle: &GroupFilesHandle,
op: &'static str,
key: &str,
result: serde_json::Value,
) -> Result<(), Box<EvalAltResult>> {
super::interceptor::run_after(
&handle.ictx,
&handle.cx,
"files",
op,
&handle.collection,
key,
None,
result,
)
}
fn register_group_list(engine: &mut RhaiEngine) {
engine.register_fn(
"list",

107
crates/executor-core/src/sdk/http.rs
View File

@@ -51,29 +51,31 @@ const MAX_REDIRECTS: i64 = 10;
const ALLOWED_OPT_KEYS: [&str; 4] = ["headers", "timeout_ms", "follow_redirects", "max_redirects"];
// `http` registers its verbs as module native fns (no handle struct), so there
// is nowhere to stash the interceptor ctx yet. The param is threaded uniformly
// from `register_all` (§9.4 M1) and consumed by M11, which adds the before/after
// hook around the outbound-request closures. Prefixed `_` until then.
// `http` registers its verbs as module native fns (no handle struct), so the
// interceptor ctx is captured into each request closure directly. Every verb
// funnels through `invoke` / `invoke_form`, which run the §9.4 before/after hook
// around `svc.request` (M11). The hook sees `service = "http"`, `op =
// "request"`, `collection = <METHOD>`, `key = <url>`; the body is never
// surfaced, so there is no data transform.
pub(super) fn register(
engine: &mut RhaiEngine,
services: &Services,
cx: Arc<SdkCallCx>,
_ictx: InterceptorCtx,
ictx: InterceptorCtx,
) {
let svc = services.http.clone();
let mut module = Module::new();
// Bodyless verbs: (url) / (url, opts).
for verb in ["get", "head"] {
register_bodyless(&mut module, verb, &svc, &cx);
register_bodyless(&mut module, verb, &svc, &cx, &ictx);
}
// Body verbs: (url) / (url, body) / (url, body, opts).
for verb in ["post", "put", "patch", "delete"] {
register_body(&mut module, verb, &svc, &cx);
register_body(&mut module, verb, &svc, &cx, &ictx);
}
register_post_form(&mut module, &svc, &cx);
register_request(&mut module, &svc, &cx);
register_post_form(&mut module, &svc, &cx, &ictx);
register_request(&mut module, &svc, &cx, &ictx);
engine.register_static_module("http", module.into());
}
@@ -83,17 +85,18 @@ fn register_bodyless(
verb: &'static str,
svc: &Arc<dyn HttpService>,
cx: &Arc<SdkCallCx>,
ictx: &InterceptorCtx,
) {
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn(verb, move |url: &str| {
invoke(&svc, &cx, verb, url, None, None)
invoke(&ictx, &svc, &cx, verb, url, None, None)
});
}
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn(verb, move |url: &str, opts: Map| {
invoke(&svc, &cx, verb, url, None, Some(&opts))
invoke(&ictx, &svc, &cx, verb, url, None, Some(&opts))
});
}
}
@@ -103,61 +106,72 @@ fn register_body(
verb: &'static str,
svc: &Arc<dyn HttpService>,
cx: &Arc<SdkCallCx>,
ictx: &InterceptorCtx,
) {
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn(verb, move |url: &str| {
invoke(&svc, &cx, verb, url, None, None)
invoke(&ictx, &svc, &cx, verb, url, None, None)
});
}
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn(verb, move |url: &str, body: Dynamic| {
invoke(&svc, &cx, verb, url, Some(body), None)
invoke(&ictx, &svc, &cx, verb, url, Some(body), None)
});
}
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn(verb, move |url: &str, body: Dynamic, opts: Map| {
invoke(&svc, &cx, verb, url, Some(body), Some(&opts))
invoke(&ictx, &svc, &cx, verb, url, Some(body), Some(&opts))
});
}
}
fn register_post_form(module: &mut Module, svc: &Arc<dyn HttpService>, cx: &Arc<SdkCallCx>) {
fn register_post_form(
module: &mut Module,
svc: &Arc<dyn HttpService>,
cx: &Arc<SdkCallCx>,
ictx: &InterceptorCtx,
) {
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn("post_form", move |url: &str, form: Map| {
invoke_form(&svc, &cx, url, &form, None)
invoke_form(&ictx, &svc, &cx, url, &form, None)
});
}
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn("post_form", move |url: &str, form: Map, opts: Map| {
invoke_form(&svc, &cx, url, &form, Some(&opts))
invoke_form(&ictx, &svc, &cx, url, &form, Some(&opts))
});
}
}
fn register_request(module: &mut Module, svc: &Arc<dyn HttpService>, cx: &Arc<SdkCallCx>) {
fn register_request(
module: &mut Module,
svc: &Arc<dyn HttpService>,
cx: &Arc<SdkCallCx>,
ictx: &InterceptorCtx,
) {
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn("request", move |method: &str, url: &str| {
invoke(&svc, &cx, method, url, None, None)
invoke(&ictx, &svc, &cx, method, url, None, None)
});
}
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn("request", move |method: &str, url: &str, body: Dynamic| {
invoke(&svc, &cx, method, url, Some(body), None)
invoke(&ictx, &svc, &cx, method, url, Some(body), None)
});
}
{
let (svc, cx) = (svc.clone(), cx.clone());
let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone());
module.set_native_fn(
"request",
move |method: &str, url: &str, body: Dynamic, opts: Map| {
invoke(&svc, &cx, method, url, Some(body), Some(&opts))
invoke(&ictx, &svc, &cx, method, url, Some(body), Some(&opts))
},
);
}
@@ -260,8 +274,9 @@ fn dispatch_body(body: Dynamic) -> Result<EncodedBody, Box<EvalAltResult>> {
Ok((Some(bytes), Some("application/json".to_string())))
}
#[allow(clippy::needless_pass_by_value)]
#[allow(clippy::needless_pass_by_value, clippy::too_many_arguments)]
fn invoke(
ictx: &InterceptorCtx,
svc: &Arc<dyn HttpService>,
cx: &Arc<SdkCallCx>,
method: &str,
@@ -271,6 +286,9 @@ fn invoke(
) -> Result<Dynamic, Box<EvalAltResult>> {
let opts = parse_opts(opts)?;
let method_uc = method.to_ascii_uppercase();
// §9.4 before-op interceptor (allow/deny only; the body is not surfaced to
// the hook). `collection` = the HTTP method, `key` = the URL.
super::interceptor::run_before(ictx, cx, "http", "request", &method_uc, url, None)?;
let bodyless = matches!(method_uc.as_str(), "GET" | "HEAD");
let (encoded, content_type) = if bodyless {
(None, None)
@@ -281,7 +299,7 @@ fn invoke(
};
let req = HttpRequest {
method: method_uc,
method: method_uc.clone(),
url: url.to_string(),
headers: opts.headers,
body: encoded,
@@ -296,11 +314,22 @@ fn invoke(
let cx = cx.clone();
block_on("http", async move { svc.request(&cx, req).await })?
};
super::interceptor::run_after(
ictx,
cx,
"http",
"request",
&method_uc,
url,
None,
serde_json::Value::Null,
)?;
Ok(response_to_dynamic(&resp))
}
#[allow(clippy::needless_pass_by_value)]
fn invoke_form(
ictx: &InterceptorCtx,
svc: &Arc<dyn HttpService>,
cx: &Arc<SdkCallCx>,
url: &str,
@@ -308,6 +337,8 @@ fn invoke_form(
opts: Option<&Map>,
) -> Result<Dynamic, Box<EvalAltResult>> {
let opts = parse_opts(opts)?;
// §9.4 before-op interceptor. `post_form` is always a POST.
super::interceptor::run_before(ictx, cx, "http", "request", "POST", url, None)?;
let mut serializer = url::form_urlencoded::Serializer::new(String::new());
for (k, v) in form {
serializer.append_pair(k.as_str(), &dyn_to_string(v));
@@ -330,6 +361,16 @@ fn invoke_form(
let cx = cx.clone();
block_on("http", async move { svc.request(&cx, req).await })?
};
super::interceptor::run_after(
ictx,
cx,
"http",
"request",
"POST",
url,
None,
serde_json::Value::Null,
)?;
Ok(response_to_dynamic(&resp))
}

66
crates/executor-core/src/sdk/pubsub.rs
View File

@@ -38,16 +38,39 @@ pub(super) fn register(
{
let svc = svc.clone();
let cx = cx.clone();
let ictx = ictx.clone();
module.set_native_fn(
"publish_durable",
move |topic: &str, message: Dynamic| -> Result<(), Box<EvalAltResult>> {
crate::sdk::emit_budget::charge_emission("pubsub::publish_durable")?;
let json = message_to_json(&message)?;
let svc = svc.clone();
let cx = cx.clone();
// §9.4 before-op interceptor (allow/deny + M4 data-transform).
let write_val = super::interceptor::run_before(
&ictx,
&cx,
"pubsub",
"publish",
topic,
"",
Some(&json),
)?
.unwrap_or(json);
let written = write_val.clone();
let svc2 = svc.clone();
let cx2 = cx.clone();
block_on("pubsub", async move {
svc.publish_durable(&cx, topic, json).await
})
svc2.publish_durable(&cx2, topic, write_val).await
})?;
super::interceptor::run_after(
&ictx,
&cx,
"pubsub",
"publish",
topic,
"",
Some(&written),
Json::Null,
)
},
);
}
@@ -112,9 +135,7 @@ pub struct GroupTopicHandle {
namespace: String,
service: Arc<dyn picloud_shared::GroupPubsubService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing: carried so `publish` can run a before/after hook in a
// later milestone (M11). Unused until then.
#[allow(dead_code)]
// §9.4: carried so shared `publish` runs the before/after hook too (M11).
ictx: InterceptorCtx,
}
@@ -125,16 +146,41 @@ fn shared_publish(
) -> Result<(), Box<EvalAltResult>> {
crate::sdk::emit_budget::charge_emission("pubsub::shared_topic")?;
let json = message_to_json(&message)?;
// §9.4 before-op interceptor also guards shared-topic publishes. The
// collection is the namespace; the subtopic is the key.
let write_val = super::interceptor::run_before(
&h.ictx,
&h.cx,
"pubsub",
"publish",
&h.namespace,
subtopic,
Some(&json),
)?
.unwrap_or(json);
let written = write_val.clone();
let service = h.service.clone();
let cx = h.cx.clone();
let namespace = h.namespace.clone();
let subtopic = subtopic.to_string();
let subtopic_owned = subtopic.to_string();
block_on("pubsub", async move {
service.publish(&cx, &namespace, &subtopic, json).await
service
.publish(&cx, &namespace, &subtopic_owned, write_val)
.await
})
// The fan-out count isn't surfaced to scripts (fire-and-forget, like the
// per-app publish).
.map(|_n: u32| ())
.map(|_n: u32| ())?;
super::interceptor::run_after(
&h.ictx,
&h.cx,
"pubsub",
"publish",
&h.namespace,
subtopic,
Some(&written),
Json::Null,
)
}
fn register_shared_publish(engine: &mut RhaiEngine) {

81
crates/executor-core/src/sdk/queue.rs
View File

@@ -43,12 +43,11 @@ pub(super) fn register(
{
let svc = svc.clone();
let cx = cx.clone();
let ictx = ictx.clone();
module.set_native_fn(
"enqueue",
move |name: &str, message: Dynamic| -> Result<(), Box<EvalAltResult>> {
crate::sdk::emit_budget::charge_emission("queue::enqueue")?;
let json = message_to_json(&message)?;
enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default())
app_enqueue(&ictx, &svc, &cx, name, &message, EnqueueOpts::default())
},
);
}
@@ -58,13 +57,12 @@ pub(super) fn register(
{
let svc = svc.clone();
let cx = cx.clone();
let ictx = ictx.clone();
module.set_native_fn(
"enqueue",
move |name: &str, message: Dynamic, opts: Map| -> Result<(), Box<EvalAltResult>> {
crate::sdk::emit_budget::charge_emission("queue::enqueue")?;
let json = message_to_json(&message)?;
let opts = parse_opts(&opts)?;
enqueue_blocking(&svc, &cx, name, json, opts)
app_enqueue(&ictx, &svc, &cx, name, &message, opts)
},
);
}
@@ -134,12 +132,42 @@ pub struct GroupQueueHandle {
collection: String,
service: Arc<dyn GroupQueueService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing: carried so `enqueue` can run a before/after hook in a
// later milestone (M10). Unused until then.
#[allow(dead_code)]
// §9.4: carried so shared `enqueue` runs the before/after hook too (M10).
ictx: InterceptorCtx,
}
/// Per-app enqueue with the §9.4 interceptor around it: charge the emission
/// budget, run the before-hook (allow/deny + M4 data-transform), enqueue the
/// (possibly rewritten) message, then run the after-hook with the new message
/// id as the result. `key` is `""` (a queue has no key). `collection` = the
/// queue name.
fn app_enqueue(
ictx: &InterceptorCtx,
svc: &Arc<dyn QueueService>,
cx: &Arc<SdkCallCx>,
name: &str,
message: &Dynamic,
opts: EnqueueOpts,
) -> Result<(), Box<EvalAltResult>> {
crate::sdk::emit_budget::charge_emission("queue::enqueue")?;
let json = message_to_json(message)?;
let write_val =
super::interceptor::run_before(ictx, cx, "queue", "enqueue", name, "", Some(&json))?
.unwrap_or(json);
let written = write_val.clone();
let id = enqueue_blocking(svc, cx, name, write_val, opts)?;
super::interceptor::run_after(
ictx,
cx,
"queue",
"enqueue",
name,
"",
Some(&written),
Json::String(id.0.to_string()),
)
}
fn group_enqueue(
h: &mut GroupQueueHandle,
message: Dynamic,
@@ -147,6 +175,18 @@ fn group_enqueue(
) -> Result<(), Box<EvalAltResult>> {
crate::sdk::emit_budget::charge_emission("queue::shared_collection")?;
let json = message_to_json(&message)?;
// §9.4 before-op interceptor also guards shared-queue enqueues.
let write_val = super::interceptor::run_before(
&h.ictx,
&h.cx,
"queue",
"enqueue",
&h.collection,
"",
Some(&json),
)?
.unwrap_or(json);
let written = write_val.clone();
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("queue: no tokio runtime available: {e}").into(),
@@ -157,12 +197,21 @@ fn group_enqueue(
let service = h.service.clone();
let cx = h.cx.clone();
let collection = h.collection.clone();
handle
.block_on(async move { service.enqueue(&cx, &collection, json, opts).await })
.map(|_id| ())
let id = handle
.block_on(async move { service.enqueue(&cx, &collection, write_val, opts).await })
.map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into()
})
})?;
super::interceptor::run_after(
&h.ictx,
&h.cx,
"queue",
"enqueue",
&h.collection,
"",
Some(&written),
Json::String(id.0.to_string()),
)
}
fn group_depth<F, Fut>(h: &mut GroupQueueHandle, f: F) -> Result<i64, Box<EvalAltResult>>
@@ -209,14 +258,15 @@ fn register_shared_queue_methods(engine: &mut RhaiEngine) {
}
/// Run the async enqueue inside the synchronous Rhai context. Mirrors
/// `pubsub::block_on` but for the queue's `Result<_, QueueError>`.
/// `pubsub::block_on` but for the queue's `Result<_, QueueError>`. Returns the
/// new message id (surfaced to the §9.4 after-hook, not to the script).
fn enqueue_blocking(
svc: &Arc<dyn QueueService>,
cx: &Arc<SdkCallCx>,
name: &str,
payload: Json,
opts: EnqueueOpts,
) -> Result<(), Box<EvalAltResult>> {
) -> Result<picloud_shared::QueueMessageId, Box<EvalAltResult>> {
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("queue: no tokio runtime available: {e}").into(),
@@ -229,7 +279,6 @@ fn enqueue_blocking(
let name = name.to_string();
handle
.block_on(async move { svc.enqueue(&cx, &name, payload, opts).await })
.map(|_id| ())
.map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into()
})